HCD-83 Enables IAuthenticator's to return own AuthenticateMessage

CHANGES.txt

@@ -1,6 +1,7 @@
 Future version (tbd)
  * Require only MODIFY permission on base when updating table with MV (STAR-564)
 Merged from 5.0:
+ * Enables IAuthenticator's to return own AuthenticateMessage (CASSANDRA-19984)
  * Disable chronicle analytics (CASSANDRA-19656)
  * Remove mocking in InternalNodeProbe spying on StorageServiceMBean (CASSANDRA-18152)
  * Fix ClassCastException  from jdk GaloisCounterMode when using JDK17 provider (CASSANDRA-18180)

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

@@ -25,6 +25,8 @@
 
 import org.apache.cassandra.exceptions.AuthenticationException;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.transport.messages.AuthenticateMessage;
 
 public interface IAuthenticator
 {
@@ -55,6 +57,17 @@ public interface IAuthenticator
      */
     void setup();
 
+    /**
+     * Allows custom authenticators to return their own {@link AuthenticateMessage} based on
+     * {@link ClientState} information. For example, this allows returning the FQCN of a driver's
+     * known authenticator (e.g. "com.datastax.bdp.cassandra.auth.DseAuthenticator") to enable
+     * SASL scheme negotiation.
+     */
+    default AuthenticateMessage getAuthenticateMessage(ClientState clientState)
+    {
+        return new AuthenticateMessage(getClass().getName());
+    }
+
     /**
      * Provide a SASL handler to perform authentication for an single connection. SASL
      * is a stateful protocol, so a new instance must be used for each authentication

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

@@ -23,6 +23,7 @@
 
 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;
@@ -127,8 +128,9 @@ else if (compression.equals("lz4"))
             clientState.setDriverVersion(options.get(DRIVER_VERSION));
         }
 
-        if (DatabaseDescriptor.getAuthenticator().requireAuthentication())
-            return new AuthenticateMessage(DatabaseDescriptor.getAuthenticator().getClass().getName());
+        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
+        if (authenticator.requireAuthentication())
+            return authenticator.getAuthenticateMessage(clientState);
         else
             return new ReadyMessage();
     }

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

@@ -0,0 +1,89 @@
+/*
+ * 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.auth;
+
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Test;
+
+import org.apache.cassandra.exceptions.AuthenticationException;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.transport.messages.AuthenticateMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+public class CustomAuthenticatorTest
+{
+    private static final String CUSTOM_AUTHENTICATOR_FQCN = "com.example.auth.CustomAuthenticator";
+
+    @Test
+    public void testCustomAuthenticator()
+    {
+        IAuthenticator authenticator = new CustomAuthenticator();
+
+        AuthenticateMessage message = authenticator.getAuthenticateMessage(ClientState.forInternalCalls());
+
+        assertThat(message.authenticator).isNotEqualTo(authenticator.getClass().getName());
+        assertThat(message.authenticator).isEqualTo(CUSTOM_AUTHENTICATOR_FQCN);
+    }
+
+    private static class CustomAuthenticator implements IAuthenticator
+    {
+        @Override
+        public boolean requireAuthentication()
+        {
+            return false;
+        }
+
+        @Override
+        public Set<? extends IResource> protectedResources()
+        {
+            return Set.of();
+        }
+
+        @Override
+        public void validateConfiguration() throws ConfigurationException {}
+
+        @Override
+        public void setup() {}
+
+        @Override
+        public AuthenticateMessage getAuthenticateMessage(ClientState clientState)
+        {
+            return new AuthenticateMessage(CUSTOM_AUTHENTICATOR_FQCN);
+        }
+
+
+        @Override
+        public SaslNegotiator newSaslNegotiator(InetAddress clientAddress)
+        {
+            return null;
+        }
+
+        @Override
+        public AuthenticatedUser legacyAuthenticate(Map<String, String> credentials) throws AuthenticationException
+        {
+            return null;
+        }
+    }
+}

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

@@ -36,9 +36,12 @@
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.transport.messages.AuthenticateMessage;
 
 import static org.apache.cassandra.auth.CassandraRoleManager.*;
 import static org.apache.cassandra.auth.PasswordAuthenticator.*;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mindrot.jbcrypt.BCrypt.hashpw;
@@ -175,4 +178,11 @@ public static void tearDown()
     {
         schemaChange("DROP KEYSPACE " + SchemaConstants.AUTH_KEYSPACE_NAME);
     }
+
+    @Test
+    public void testDefaultAuthenticateMessage()
+    {
+        AuthenticateMessage authenticateMessage = authenticator.getAuthenticateMessage(null);
+        assertThat(authenticateMessage.authenticator).isEqualTo(PasswordAuthenticator.class.getName());
+    }
 }