Allow CQL client certificate authentication to work without sending an AUTHENTICATE request

patch by Andy Tolbert; reviewed by Abe Ratnofsky, Dinesh Joshi, Francisco Guerrero, Jyothsna Konisa for CASSANDRA-18857

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Allow CQL client certificate authentication to work without sending an AUTHENTICATE request (CASSANDRA-18857)
  * Extend nodetool tpstats and system_views.thread_pools with detailed pool parameters (CASSANDRA-19289) 
  * Remove dependency on Sigar in favor of OSHI (CASSANDRA-16565)
  * Simplify the bind marker and Term logic (CASSANDRA-18813)

src/java/org/apache/cassandra/auth/AuthCacheService.java

@@ -77,4 +77,18 @@ public class AuthCacheService
             Roles.init();
         }
     }
+
+    /**
+     * Invalidates all registered caches, which is useful for tests that make changes to roles, identities, cidr
+     * permissions and so on.
+     */
+    @VisibleForTesting
+    public synchronized void invalidateCaches()
+    {
+        logger.info("Bulk invalidating {} auth cache(s)", caches.size());
+        for (AuthCache<?, ?> cache : caches)
+        {
+            cache.invalidate();
+        }
+    }
 }

src/java/org/apache/cassandra/auth/IAuthenticator.java

@@ -28,11 +28,31 @@ import org.apache.cassandra.exceptions.ConfigurationException;
 public interface IAuthenticator
 {
     /**
-     * Whether or not the authenticator requires explicit login.
+     * Whether the authenticator requires explicit login.
      * If false will instantiate user with AuthenticatedUser.ANONYMOUS_USER.
      */
     boolean requireAuthentication();
 
+    /**
+     * Whether the authenticator supports 'early' authentication, meaning that it can be authenticated without
+     * the need to send an AUTHENTICATE request to the client after receiveing a STARTUP message from the client.
+     * <p>
+     * An example use case of this would be if the client could be authenticated using certificates present
+     * on a TLS-encrypted connection, as is done in {@link MutualTlsAuthenticator}.  In this case, an AUTHENTICATE
+     * message is not needed because the client can be identified by information present in its provided certificate.
+     * <p>
+     * If an authenticator supports requires early authentication, this should be <code>true</code>;
+     * otherwise if a client cannot authenticate using its connection details (e.g. a certificate), an AUTHENTICATE
+     * will be sent to the client. This may confuse a driver implementation into thinking an authenticator is needed,
+     * when instead the real issue is that something is missing on the connection, such as a certificate.
+     * <p>
+     * If <code>false</code> (default behavior), an AUTHENTICATE request will be sent to the client.
+     */
+    default boolean supportsEarlyAuthentication()
+    {
+        return false;
+    }
+
      /**
      * Set of resources that should be made inaccessible to users and only accessible internally.
      *
@@ -145,5 +165,19 @@ public interface IAuthenticator
          * @throws AuthenticationException
          */
         public AuthenticatedUser getAuthenticatedUser() throws AuthenticationException;
+
+        /**
+         * Whether it is determined that an AUTHENTICATE message should be sent to properly negotiate.
+         * This is used in conjunction with {@link #supportsEarlyAuthentication()} in determining if authentication
+         * can be done early, for example if the client can be authenticated using client certificates when handling
+         * a STARTUP message.
+         * <p>
+         * If <code>true</code> (default behavior), an AUTHENTICATE message will be sent in response to a STARTUP,
+         * otherwise {@link #evaluateResponse(byte[])} will be called with an empty byte array when handling STARTUP.
+         */
+        default boolean shouldSendAuthenticateMessage()
+        {
+            return true;
+        }
     }
 }

src/java/org/apache/cassandra/auth/MutualTlsAuthenticator.java

@@ -90,6 +90,12 @@ public class MutualTlsAuthenticator implements IAuthenticator
         return true;
     }
 
+    @Override
+    public boolean supportsEarlyAuthentication()
+    {
+        return true;
+    }
+
     @Override
     public Set<? extends IResource> protectedResources()
     {
@@ -149,6 +155,12 @@ public class MutualTlsAuthenticator implements IAuthenticator
             return null;
         }
 
+        @Override
+        public boolean shouldSendAuthenticateMessage()
+        {
+            return false;
+        }
+
         @Override
         public boolean isComplete()
         {
@@ -158,6 +170,11 @@ public class MutualTlsAuthenticator implements IAuthenticator
         @Override
         public AuthenticatedUser getAuthenticatedUser() throws AuthenticationException
         {
+            if (clientCertificateChain == null || clientCertificateChain.length == 0)
+            {
+                throw new AuthenticationException("No certificate present on connection");
+            }
+
             if (!certificateValidator.isValidCertificate(clientCertificateChain))
             {
                 String message = "Invalid or not supported certificate";

src/java/org/apache/cassandra/auth/MutualTlsWithPasswordFallbackAuthenticator.java

@@ -46,13 +46,21 @@ public class MutualTlsWithPasswordFallbackAuthenticator extends PasswordAuthenti
         mutualTlsAuthenticator.setup();
     }
 
+    @Override
+    public boolean supportsEarlyAuthentication()
+    {
+        return true;
+    }
+
     @Override
     public SaslNegotiator newSaslNegotiator(InetAddress clientAddress, Certificate[] certificates)
     {
         if (certificates == null || certificates.length == 0)
         {
+            // If no certificates present, fallback to PasswordAuthentication
             return newSaslNegotiator(clientAddress);
         }
+        // Otherwise attempt to authenticate using the client-provided certificate.
         return mutualTlsAuthenticator.newSaslNegotiator(clientAddress, certificates);
     }
 

src/java/org/apache/cassandra/transport/SimpleClient.java

@@ -416,12 +416,13 @@ public class SimpleClient implements Closeable
                     }
                     break;
                 case SUPPORTED:
+                case ERROR:
                     // just pass through
                     results.add(response);
                     break;
                 default:
                     throw new ProtocolException(String.format("Unexpected %s response expecting " +
-                                                              "READY, AUTHENTICATE or SUPPORTED",
+                                                              "READY, AUTHENTICATE, ERROR or SUPPORTED",
                                                               response.header.type));
             }
         }

src/java/org/apache/cassandra/transport/messages/AuthResponse.java

@@ -20,11 +20,6 @@ package org.apache.cassandra.transport.messages;
 import java.nio.ByteBuffer;
 
 import io.netty.buffer.ByteBuf;
-import org.apache.cassandra.auth.AuthEvents;
-import org.apache.cassandra.auth.AuthenticatedUser;
-import org.apache.cassandra.auth.IAuthenticator;
-import org.apache.cassandra.exceptions.AuthenticationException;
-import org.apache.cassandra.metrics.ClientMetrics;
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.transport.*;
 
@@ -71,29 +66,16 @@ public class AuthResponse extends Message.Request
     @Override
     protected Response execute(QueryState queryState, long queryStartNanoTime, boolean traceRequest)
     {
-        try
+        return AuthUtil.handleLogin(connection, queryState, token, (negotiationComplete, challenge) ->
         {
-            IAuthenticator.SaslNegotiator negotiator = ((ServerConnection) connection).getSaslNegotiator(queryState);
-            byte[] challenge = negotiator.evaluateResponse(token);
-            if (negotiator.isComplete())
+            if (negotiationComplete)
             {
-                AuthenticatedUser user = negotiator.getAuthenticatedUser();
-                queryState.getClientState().login(user);
-                ClientMetrics.instance.markAuthSuccess();
-                AuthEvents.instance.notifyAuthSuccess(queryState);
-                // authentication is complete, send a ready message to the client
                 return new AuthSuccess(challenge);
             }
             else
             {
                 return new AuthChallenge(challenge);
             }
-        }
-        catch (AuthenticationException e)
-        {
-            ClientMetrics.instance.markAuthFailure();
-            AuthEvents.instance.notifyAuthFailure(queryState, e);
-            return ErrorMessage.fromException(e);
-        }
+        });
     }
 }

src/java/org/apache/cassandra/transport/messages/AuthUtil.java

@@ -0,0 +1,100 @@
+/*
+ * 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.cassandra.transport.messages;
+
+import java.util.function.BiFunction;
+
+import org.apache.cassandra.auth.AuthEvents;
+import org.apache.cassandra.auth.AuthenticatedUser;
+import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.exceptions.AuthenticationException;
+import org.apache.cassandra.metrics.ClientMetrics;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.Connection;
+import org.apache.cassandra.transport.Message.Response;
+import org.apache.cassandra.transport.ServerConnection;
+
+public final class AuthUtil
+{
+
+    /**
+     * Attempts to authenticate a user on the current connection by negotiating with the underlying
+     * {@link org.apache.cassandra.auth.IAuthenticator.SaslNegotiator} using the given state of the connection
+     * (e.g. client certificates), the provided query state and the provided token.
+     * <p>
+     * If unsuccessful an {@link ErrorMessage} is returned and the authentication failure is recorded.
+     * <p>
+     * If negotiation is complete, a successful login is recorded and the given function is called with
+     * <code>true</code> and the challenge returned from
+     * {@link org.apache.cassandra.auth.IAuthenticator.SaslNegotiator#evaluateResponse(byte[])}.
+     * <p>
+     * If negotiation is incomplete, <code>false</code> and the challenge is passed to the given function.
+     *
+     * @param connection                      The connection to authenticate
+     * @param queryState                      The current query state
+     * @param token                           The token provided in an {@link AuthResponse} from the client (or empty
+     *                                        if not handling an AuthResponse).
+     * @param messageToSendBasedOnNegotiation Determines what response to return on based on whether sasl negotiation
+     *                                        is complete (1st parameter) and the challenege token returned from the
+     *                                        negotiator (2nd parameter).
+     * @return the response to send back to the client.
+     */
+    static Response handleLogin(Connection connection, QueryState queryState, byte[] token,
+                                BiFunction<Boolean, byte[], Response> messageToSendBasedOnNegotiation)
+    {
+        try
+        {
+            // client-side timeout can disconnect while sitting in auth executor queue so (client default 12s)
+            // discard if connection closed anyway
+            if (!connection.channel().isActive())
+            {
+                throw new AuthenticationException("Auth check after connection closed");
+            }
+            IAuthenticator.SaslNegotiator negotiator = ((ServerConnection) connection).getSaslNegotiator(queryState);
+            byte[] challenge = negotiator.evaluateResponse(token);
+            if (negotiator.isComplete())
+            {
+                AuthenticatedUser user = negotiator.getAuthenticatedUser();
+                queryState.getClientState().login(user);
+                ClientMetrics.instance.markAuthSuccess();
+                AuthEvents.instance.notifyAuthSuccess(queryState);
+                // authentication is complete, complete the authentication flow.
+                return messageToSendBasedOnNegotiation.apply(true, challenge);
+            }
+            else
+            {
+                // authentication is incomplete, continue the authentication flow.
+                return messageToSendBasedOnNegotiation.apply(false, challenge);
+            }
+        }
+        catch (AuthenticationException e)
+        {
+            ClientMetrics.instance.markAuthFailure();
+            AuthEvents.instance.notifyAuthFailure(queryState, e);
+            return ErrorMessage.fromException(e);
+        }
+    }
+
+    /**
+     * The class must not be instantiated.
+     */
+    private AuthUtil()
+    {
+    }
+}

src/java/org/apache/cassandra/transport/messages/StartupMessage.java

@@ -22,6 +22,7 @@ import java.util.Map;
 
 import io.netty.buffer.ByteBuf;
 
+import org.apache.cassandra.auth.IAuthenticator;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
@@ -59,6 +60,8 @@ public class StartupMessage extends Message.Request
         }
     };
 
+    private static final byte[] EMPTY_CLIENT_RESPONSE = new byte[0];
+
     public final Map<String, String> options;
 
     public StartupMessage(Map<String, String> options)
@@ -118,8 +121,35 @@ public class StartupMessage extends Message.Request
             clientState.setDriverVersion(options.get(DRIVER_VERSION));
         }
 
-        if (DatabaseDescriptor.getAuthenticator().requireAuthentication())
+        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
+        if (authenticator.requireAuthentication())
+        {
+            // If the authenticator supports early authentication, attempt to authenticate.
+            if (authenticator.supportsEarlyAuthentication())
+            {
+                IAuthenticator.SaslNegotiator negotiator = ((ServerConnection) connection).getSaslNegotiator(state);
+                // If the negotiator determines that sending an authenticate message is not necessary, attempt to authenticate here,
+                // otherwise, send an Authenticate message to begin the traditional authentication flow.
+                if (!negotiator.shouldSendAuthenticateMessage())
+                {
+                    // Attempt to authenticate the user.
+                    return AuthUtil.handleLogin(connection, state, EMPTY_CLIENT_RESPONSE, (negotiationComplete, challenge) ->
+                    {
+                        if (negotiationComplete)
+                        {
+                            // Authentication was successful, proceed.
+                            return new ReadyMessage();
+                        } else
+                        {
+                            // It's expected that any negotiator that requires a challenge will likely not support early
+                            // authentication, in this case we can just go through the traditional auth flow.
                             return new AuthenticateMessage(DatabaseDescriptor.getAuthenticator().getClass().getName());
+                        }
+                    });
+                }
+            }
+            return new AuthenticateMessage(DatabaseDescriptor.getAuthenticator().getClass().getName());
+        }
         else
             return new ReadyMessage();
     }

test/unit/org/apache/cassandra/auth/AuthTestUtils.java

@@ -33,6 +33,8 @@ import java.util.Collection;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeoutException;
 
+import com.google.common.base.Charsets;
+
 import org.apache.cassandra.auth.jmx.AuthorizationProxy;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.CIDR;
@@ -239,6 +241,27 @@ public class AuthTestUtils
         }
     }
 
+    public static class LocalMutualTlsWithPasswordFallbackAuthenticator extends MutualTlsWithPasswordFallbackAuthenticator
+    {
+
+        public LocalMutualTlsWithPasswordFallbackAuthenticator(Map<String, String> parameters)
+        {
+            super(parameters);
+        }
+
+        @Override
+        ResultMessage.Rows select(SelectStatement statement, QueryOptions options)
+        {
+            return statement.executeLocally(QueryState.forInternalCalls(), options);
+        }
+
+        @Override
+        UntypedResultSet process(String query, ConsistencyLevel cl)
+        {
+            return QueryProcessor.executeInternal(query);
+        }
+    }
+
     public static class NoAuthSetupAuthorizationProxy extends AuthorizationProxy
     {
         public NoAuthSetupAuthorizationProxy()
@@ -363,9 +386,38 @@ public class AuthTestUtils
     }
 
     public static void initializeIdentityRolesTable(final String identity) throws IOException, TimeoutException
+    {
+        truncateIdentityRolesTable();
+        addIdentityToRole(identity, "readonly_user");
+    }
+
+    public static void truncateIdentityRolesTable() throws IOException, TimeoutException
     {
         StorageService.instance.truncate(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.IDENTITY_TO_ROLES);
-        String insertQuery = "Insert into %s.%s (identity, role) values ('%s', 'readonly_user');";
-        QueryProcessor.process(String.format(insertQuery, SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.IDENTITY_TO_ROLES, identity), ConsistencyLevel.ONE);
+    }
+
+    public static void addIdentityToRole(final String identity, final String role)
+    {
+        String insertQuery = "INSERT INTO %s.%s (identity, role) VALUES ('%s', '%s');";
+        QueryProcessor.process(String.format(insertQuery, SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.IDENTITY_TO_ROLES, identity, role), ConsistencyLevel.ONE);
+    }
+
+    /**
+     * Convenience method for producing a username:password token expected by {@link PasswordAuthenticator}
+     * @param username user to encode
+     * @param password password to encode
+     * @return Byte array formatted as 0-byte, username, 0-byte, password
+     */
+    public static byte[] getToken(String username, String password)
+    {
+        byte[] usernameBytes = username.getBytes(Charsets.UTF_8);
+        byte[] passwordBytes = password.getBytes(Charsets.UTF_8);
+        // Format of the token sent is 0-byte, username, 0-bytes, password
+        byte[] token = new byte[usernameBytes.length + passwordBytes.length + 2];
+        token[0] = 0;
+        System.arraycopy(usernameBytes, 0, token, 1, usernameBytes.length);
+        token[usernameBytes.length + 1] = 0;
+        System.arraycopy(passwordBytes, 0, token, usernameBytes.length + 2, passwordBytes.length);
+        return token;
     }
 }

test/unit/org/apache/cassandra/cql3/CQLTester.java

@@ -97,6 +97,7 @@ import org.apache.cassandra.Util;
 import org.apache.cassandra.auth.AuthCacheService;
 import org.apache.cassandra.auth.AuthSchemaChangeListener;
 import org.apache.cassandra.auth.AuthTestUtils;
+import org.apache.cassandra.auth.IAuthenticator;
 import org.apache.cassandra.auth.IRoleManager;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.config.CassandraRelevantProperties;
@@ -518,7 +519,12 @@ public abstract class CQLTester
 
     protected static void requireAuthentication()
     {
-        DatabaseDescriptor.setAuthenticator(new AuthTestUtils.LocalPasswordAuthenticator());
+        requireAuthentication(new AuthTestUtils.LocalPasswordAuthenticator());
+    }
+
+    protected static void requireAuthentication(final IAuthenticator authenticator)
+    {
+        DatabaseDescriptor.setAuthenticator(authenticator);
         DatabaseDescriptor.setAuthorizer(new AuthTestUtils.LocalCassandraAuthorizer());
         DatabaseDescriptor.setNetworkAuthorizer(new AuthTestUtils.LocalCassandraNetworkAuthorizer());
         DatabaseDescriptor.setCIDRAuthorizer(new AuthTestUtils.LocalCassandraCIDRAuthorizer());
@@ -530,6 +536,7 @@ public abstract class CQLTester
             public void setup()
             {
                 loadRoleStatement();
+                loadIdentityStatement();
                 QueryProcessor.executeInternal(createDefaultRoleQuery());
             }
         };
@@ -547,6 +554,27 @@ public abstract class CQLTester
         AuthCacheService.initializeAndRegisterCaches();
     }
 
+    /**
+     * Configures the server to require client encryption for CQL.  Useful for tests which exercise TLS specific
+     * behavior.
+     * <p>
+     * Note to use this appropriately, {@link #requireNetwork} should be given a server configurator configured
+     * with {@link Server.Builder#withTlsEncryptionPolicy(EncryptionOptions.TlsEncryptionPolicy)} using
+     * {@link org.apache.cassandra.config.EncryptionOptions.TlsEncryptionPolicy#ENCRYPTED}.
+     */
+    protected static void requireNativeProtocolClientEncryption()
+    {
+        DatabaseDescriptor.updateNativeProtocolEncryptionOptions((encryptionOptions ->
+                encryptionOptions.withEnabled(true)
+                        .withKeyStore("test/conf/cassandra_ssl_test.keystore")
+                        .withKeyStorePassword("cassandra")
+                        .withTrustStore("test/conf/cassandra_ssl_test.truststore")
+                        .withTrustStorePassword("cassandra")
+                        .withRequireEndpointVerification(false)
+                        .withRequireClientAuth(EncryptionOptions.ClientAuth.OPTIONAL)));
+    }
+
+
     /**
      *  Initialize Native Transport for test that need it.
      */
@@ -2205,7 +2233,8 @@ public abstract class CQLTester
     }
 
     @FunctionalInterface
-    public interface CheckedFunction {
+    public interface CheckedFunction
+    {
         void apply() throws Throwable;
     }
 

test/unit/org/apache/cassandra/transport/AuthenticationTest.java

@@ -0,0 +1,121 @@
+/*
+ * 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.cassandra.transport;
+
+import java.io.IOException;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.exceptions.AuthenticationException;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.messages.AuthResponse;
+import org.apache.cassandra.transport.messages.AuthSuccess;
+import org.apache.cassandra.transport.messages.AuthenticateMessage;
+import org.apache.cassandra.transport.messages.ErrorMessage;
+import org.apache.cassandra.transport.messages.StartupMessage;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.apache.cassandra.auth.AuthTestUtils.getToken;
+import static org.apache.cassandra.transport.messages.StartupMessage.CQL_VERSION;
+import static org.junit.Assert.fail;
+import static org.psjava.util.AssertStatus.assertTrue;
+
+/**
+ * Verifies that {@link StartupMessage#execute(QueryState, long)} will prompt a client for authentication if an
+ * authenticator is configured.  For this test {@link org.apache.cassandra.auth.PasswordAuthenticator} is used.
+ */
+public class AuthenticationTest extends CQLTester
+{
+
+    @BeforeClass
+    public static void setup()
+    {
+        requireNetwork();
+        requireAuthentication();
+    }
+
+    @Before
+    public void initNetwork()
+    {
+        reinitializeNetwork();
+    }
+
+    @Test
+    public void testSuccessfulAuth()
+    {
+        testAuthResponse((client) -> {
+            AuthResponse authResponse = new AuthResponse(getToken("cassandra", "cassandra"));
+            Message.Response authResponseResponse = client.execute(authResponse);
+
+            if (!(authResponseResponse instanceof AuthSuccess))
+            {
+                fail("Expected an AUTH_SUCCESS in response to a AUTH_RESPONSE with default credentials, got: " + authResponseResponse);
+            }
+        });
+    }
+
+    @Test
+    public void testUnsuccessfulAuth()
+    {
+        testAuthResponse((client) -> {
+            AuthResponse authResponse = new AuthResponse(getToken("cassandra", "badpw"));
+            Message.Response response = client.execute(authResponse, false);
+
+            if (response instanceof ErrorMessage)
+            {
+                ErrorMessage errorMessage = (ErrorMessage) response;
+                assertTrue(errorMessage.error instanceof AuthenticationException, "Expected an AuthenticationException, got: " + errorMessage.error);
+            }
+            else
+            {
+                fail("Expected an ErrorMessage but got: " + response);
+            }
+        });
+    }
+
+    public void testAuthResponse(Consumer<SimpleClient> testFn)
+    {
+        SimpleClient.Builder builder = SimpleClient.builder(nativeAddr.getHostAddress(), nativePort);
+        try (SimpleClient client = builder.build())
+        {
+            client.establishConnection();
+
+            // Send a StartupMessage
+            StartupMessage startup = new StartupMessage(ImmutableMap.of(CQL_VERSION, QueryProcessor.CQL_VERSION.toString()));
+            Message.Response startupResponse = client.execute(startup);
+
+            if (!(startupResponse instanceof AuthenticateMessage))
+            {
+                fail("Expected an AUTHENTICATE in response to a STARTUP, got: " + startupResponse);
+            }
+
+            testFn.accept(client);
+        }
+        catch (IOException e)
+        {
+            fail("Error establishing connection");
+        }
+    }
+}

test/unit/org/apache/cassandra/transport/EarlyAuthenticationTest.java

@@ -0,0 +1,195 @@
+/*
+ * 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.cassandra.transport;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.auth.AuthCacheService;
+import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.auth.MutualTlsAuthenticator;
+import org.apache.cassandra.auth.SpiffeCertificateValidator;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.exceptions.AuthenticationException;
+import org.apache.cassandra.transport.messages.ErrorMessage;
+import org.apache.cassandra.transport.messages.ReadyMessage;
+import org.apache.cassandra.transport.messages.StartupMessage;
+
+import static org.apache.cassandra.auth.AuthTestUtils.addIdentityToRole;
+import static org.apache.cassandra.auth.AuthTestUtils.truncateIdentityRolesTable;
+import static org.apache.cassandra.transport.messages.StartupMessage.CQL_VERSION;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.psjava.util.AssertStatus.assertTrue;
+
+/**
+ * Verifies the behavior of Cassandra given an authenticator ({@link MutualTlsAuthenticator}) returning
+ * {@link IAuthenticator#supportsEarlyAuthentication()} as <code>true</code>.
+ */
+public class EarlyAuthenticationTest extends CQLTester
+{
+
+    // configures server with client encryption enabled.
+    static final Consumer<Server.Builder> serverConfigurator = server -> server.withTlsEncryptionPolicy(EncryptionOptions.TlsEncryptionPolicy.ENCRYPTED);
+
+    static final Map<String, String> authenticatorParams = ImmutableMap.of("validator_class_name", SpiffeCertificateValidator.class.getSimpleName());
+
+    // identity present in the client cert being used.
+    static final String spiffeIdentity = "spiffe://test.cassandra.apache.org/unitTest/mtls";
+
+    @BeforeClass
+    public static void setup()
+    {
+        setupWithAuthenticator(new MutualTlsAuthenticator(authenticatorParams));
+    }
+
+    static void setupWithAuthenticator(IAuthenticator authenticator)
+    {
+        requireNativeProtocolClientEncryption();
+        requireNetwork(serverConfigurator, cluster -> {
+        });
+        requireAuthentication(authenticator);
+    }
+
+    @Before
+    public void initNetwork() throws IOException, TimeoutException
+    {
+        truncateIdentityRolesTable();
+        AuthCacheService.instance.invalidateCaches();
+        AuthCacheService.instance.warmCaches();
+        reinitializeNetwork(serverConfigurator, cluster -> {
+        });
+    }
+
+    private EncryptionOptions clientEncryptionOptions(boolean presentClientCertificate)
+    {
+        // To regenerate:
+        // 1. generate keystore
+        //    keytool -genkeypair -keystore test/conf/cassandra_ssl_test_spiffe.keystore -validity 100000 -keyalg RSA -dname "CN=Apache Cassandra, OU=ssl_test, O=Unknown, L=Unknown, ST=Unknown, C=Unknown" -keypass cassandra -storepass cassandra -alias spiffecert -ext SAN=URI:spiffe://test.cassandra.apache.org/unitTest/mtls -storetype jks
+        // 2. export cert
+        //    keytool -export -alias spiffecert -file spiffecert.cer -keystore test/conf/cassandra_ssl_test_spiffe.keystore
+        // 3. import cert into truststore
+        //    keytool -import -v -trustcacerts -alias spiffecert -file spiffecert.cer -keystore test/conf/cassandra_ssl_test.truststore
+        EncryptionOptions encryptionOptions = new EncryptionOptions()
+                                              .withEnabled(true)
+                                              .withRequireClientAuth(EncryptionOptions.ClientAuth.OPTIONAL)
+                                              .withTrustStore("test/conf/cassandra_ssl_test.truststore")
+                                              .withTrustStorePassword("cassandra")
+                                              .withSslContextFactory(new ParameterizedClass(SimpleClientSslContextFactory.class.getName()));
+
+        if (presentClientCertificate)
+        {
+            encryptionOptions = encryptionOptions.withKeyStore("test/conf/cassandra_ssl_test_spiffe.keystore")
+                             .withStoreType("JKS")
+                             .withKeyStorePassword("cassandra");
+        }
+
+        return new EncryptionOptions(encryptionOptions);
+    }
+
+    @Test
+    public void testEarlyAuthSuccess()
+    {
+        // given server is configured with a Mutual TLS Authenticator and the identity in the client's keystore is bound to cassandra.
+        addIdentityToRole(spiffeIdentity, "cassandra");
+
+        // when connecting, we expect to get a 'READY' message after sending a 'STARTUP' as MutualTlsAuthenticator
+        // supports early authentication and the client presented a cert with an identity that was bound to a role.
+        testStartupResponse(true, startupResponse -> {
+            if (!(startupResponse instanceof ReadyMessage))
+            {
+                fail("Expected an READY in response to a STARTUP, got: " + startupResponse);
+            }
+        });
+    }
+
+    @Test
+    public void testEarlyAuthFailureLoginDisallowed()
+    {
+        // given server is configured with a Mutual TLS Authenticator and the identity in the client's keystore is bound
+        // to a role that is not permitted to log in.
+        // when connecting, we expect an 'ERROR' message.
+        addIdentityToRole(spiffeIdentity, "readonly_user");
+        testStartupResponse(true, expectAuthenticationError(String.format("readonly_user is not permitted to log in", spiffeIdentity)));
+    }
+
+    @Test
+    public void testEarlyAuthFailureMissingIdentity()
+    {
+        // given server is configured with a Mutual TLS Authenticator, but no identities are bound to roles.
+        // when connecting, we expect an 'ERROR' message.
+        testStartupResponse(true, expectAuthenticationError(String.format("Certificate identity '%s' not authorized", spiffeIdentity)));
+    }
+
+    @Test
+    public void testNoClientCertificatePresented()
+    {
+        // given server is configured with a Mutual TLS Authenticator, but no client certificate is presented.
+        // when connecting, we expect an 'ERROR' message.
+        testStartupResponse(false, expectAuthenticationError("No certificate present on connection"));
+    }
+
+    public void testStartupResponse(boolean presentClientCertificate, Consumer<Message.Response> testFn)
+    {
+        SimpleClient.Builder builder = SimpleClient.builder(nativeAddr.getHostAddress(), nativePort)
+                                                   .encryption(clientEncryptionOptions(presentClientCertificate));
+        try (SimpleClient client = builder.build())
+        {
+            client.establishConnection();
+
+            // Send a StartupMessage and pass the response to the handling function
+            StartupMessage startup = new StartupMessage(ImmutableMap.of(CQL_VERSION, QueryProcessor.CQL_VERSION.toString()));
+            Message.Response startupResponse = client.execute(startup, false);
+            testFn.accept(startupResponse);
+        }
+        catch (IOException e)
+        {
+            fail("Error establishing connection");
+        }
+    }
+
+    public Consumer<Message.Response> expectAuthenticationError(final String expectedMessage)
+    {
+        return startupResponse -> {
+            if (startupResponse instanceof ErrorMessage)
+            {
+                ErrorMessage errorMessage = (ErrorMessage) startupResponse;
+                assertTrue(errorMessage.error instanceof AuthenticationException, "Expected an AuthenticationException, got: " + errorMessage.error);
+                AuthenticationException authException = (AuthenticationException) errorMessage.error;
+                assertEquals(expectedMessage, authException.getMessage());
+            }
+            else
+            {
+                fail("Expected an ErrorMessage but got: " + startupResponse);
+            }
+        };
+    }
+
+}
+

test/unit/org/apache/cassandra/transport/MutualTlsWithPasswordFallbackAuthenticatorEarlyAuthenticationTest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.cassandra.transport;
+
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.auth.AuthTestUtils;
+import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.transport.messages.AuthenticateMessage;
+
+import static org.junit.Assert.fail;
+
+/**
+ * A variant of {@link EarlyAuthenticationTest} that configures
+ * {@link org.apache.cassandra.auth.MutualTlsWithPasswordFallbackAuthenticator}
+ * as the configured authenticator.
+ * <p>
+ * This authenticator has an interesting property such that its underlying
+ * {@link IAuthenticator.SaslNegotiator#shouldSendAuthenticateMessage()} will only return false if a client
+ * certificate is present, otherwise it will return true, which should cause Cassandra to return an AUTHENTICATE
+ * message in response to a STARTUP request sent by a client which will provoke the normal authentication flow.
+ */
+public class MutualTlsWithPasswordFallbackAuthenticatorEarlyAuthenticationTest extends EarlyAuthenticationTest
+{
+
+    @BeforeClass
+    public static void setup()
+    {
+        setupWithAuthenticator(new AuthTestUtils.LocalMutualTlsWithPasswordFallbackAuthenticator(authenticatorParams));
+    }
+
+    @Test
+    @Override
+    public void testNoClientCertificatePresented()
+    {
+        /*
+         * given server is configured with a Fallback password authenticator in that it supports early certificate
+         * authentication, but the sasl negotiator is determined based on the presence of client certificates
+         *
+         * When connecting without a client certificate, we expect the server to prompt us to authenticate.
+         */
+        testStartupResponse(false, startupResponse -> {
+            if (!(startupResponse instanceof AuthenticateMessage))
+            {
+                fail("Expected an AUTHENTICATE in response to a STARTUP, got: " + startupResponse);
+            }
+        });
+    }
+}

test/unit/org/apache/cassandra/transport/SimpleClientSslContextFactory.java

@@ -0,0 +1,68 @@
+/*
+ * 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.cassandra.transport;
+
+import java.util.Map;
+import javax.net.ssl.SSLException;
+
+import io.netty.handler.ssl.CipherSuiteFilter;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import org.apache.cassandra.config.EncryptionOptions;
+import org.apache.cassandra.security.FileBasedSslContextFactory;
+
+/**
+ * A custom implementation of {@link FileBasedSslContextFactory} to be used by tests utilizing {@link SimpleClient}.
+ * <p>
+ * Provides a subtly different implementation of {@link #createNettySslContext(EncryptionOptions.ClientAuth, SocketType, CipherSuiteFilter)}
+ * that only configures an {@link SslContext} for clients and most importantly only configures a key manager if an
+ * outbound keystore is configured, where the existing implementation always does this.  This is useful for tests
+ * that try to create a client that uses encryption but does not provide a certificate.
+ */
+public class SimpleClientSslContextFactory extends FileBasedSslContextFactory
+{
+
+    public SimpleClientSslContextFactory(Map<String, Object> parameters)
+    {
+        super(parameters);
+    }
+
+    @Override
+    public SslContext createNettySslContext(EncryptionOptions.ClientAuth clientAuth, SocketType socketType,
+                                            CipherSuiteFilter cipherFilter) throws SSLException
+    {
+        SslContextBuilder builder = SslContextBuilder.forClient();
+        // only provide a client certificate if keystore is present.
+        if (hasOutboundKeystore())
+        {
+            builder.keyManager(buildOutboundKeyManagerFactory());
+        }
+
+        builder.sslProvider(getSslProvider())
+               .protocols(getAcceptedProtocols())
+               .trustManager(buildTrustManagerFactory());
+
+        // only set the cipher suites if the operator has explicity configured values for it; else, use the default
+        // for each ssl implemention (jdk or openssl)
+        if (cipher_suites != null && !cipher_suites.isEmpty())
+            builder.ciphers(cipher_suites, cipherFilter);
+
+        return builder.build();
+    }
+}