Merge branch 'cassandra-5.0' into trunk

* cassandra-5.0:
  Enables IAuthenticator's to return own AuthenticateMessage

CHANGES.txt

@@ -86,6 +86,7 @@
  * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
  * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
 Merged from 5.0:
+ * Enables IAuthenticator's to return own AuthenticateMessage (CASSANDRA-19984)
  * Use ParameterizedClass for all auth-related implementations (CASSANDRA-19946)
  * Correct out-of-date metrics and configuration documentation for SAI (CASSANDRA-19898)
  * Make configuration entries in memtable section order-independent (CASSANDRA-19906)

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

@@ -28,6 +28,8 @@ import javax.annotation.Nonnull;
 
 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
 {
@@ -78,6 +80,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

@@ -143,12 +143,12 @@ public class StartupMessage extends Message.Request
                         {
                             // 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 authenticator.getAuthenticateMessage(clientState);
                         }
                     });
                 }
             }
-            return new AuthenticateMessage(DatabaseDescriptor.getAuthenticator().getClass().getName());
+            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

@@ -40,6 +40,7 @@ import org.apache.cassandra.exceptions.AuthenticationException;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.transport.messages.AuthenticateMessage;
 
 import static org.apache.cassandra.auth.AuthTestUtils.ALL_ROLES;
 import static org.apache.cassandra.auth.CassandraRoleManager.DEFAULT_SUPERUSER_PASSWORD;
@@ -47,6 +48,7 @@ import static org.apache.cassandra.auth.CassandraRoleManager.getGensaltLogRounds
 import static org.apache.cassandra.auth.PasswordAuthenticator.SaslNegotiator;
 import static org.apache.cassandra.auth.PasswordAuthenticator.checkpw;
 import static org.apache.cassandra.config.CassandraRelevantProperties.AUTH_BCRYPT_GENSALT_LOG2_ROUNDS;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -198,4 +200,11 @@ public class PasswordAuthenticatorTest extends CQLTester
         Map<String, String> cacheEntries = authenticator.bulkLoader().get();
         assertTrue(cacheEntries.isEmpty());
     }
+
+    @Test
+    public void testDefaultAuthenticateMessage()
+    {
+        AuthenticateMessage authenticateMessage = authenticator.getAuthenticateMessage(null);
+        assertThat(authenticateMessage.authenticator).isEqualTo(PasswordAuthenticator.class.getName());
+    }
 }