CEP-50/CASS-21334: Server negotiates authenticator per client connection

conf/cassandra.yaml

@@ -191,10 +191,12 @@ batchlog_replay_throttle: 1024KiB
 #    ...
 #
 # - AllowAllAuthenticator performs no checks - set it to disable authentication.
-# - PasswordAuthenticator relies on username/password pairs to authenticate
-#   users. It keeps usernames and hashed passwords in system_auth.roles table.
-#   Please increase system_auth keyspace replication factor if you use this authenticator.
-#   If using PasswordAuthenticator, CassandraRoleManager must also be used (see below)
+# - PasswordAuthenticator relies on username/password pairs to authenticate users. It keeps usernames and
+#     hashed passwords in system_auth.roles table. Please increase system_auth keyspace replication factor
+#     if you use this authenticator.
+#     If using PasswordAuthenticator, CassandraRoleManager must also be used (see below).
+# - MutualTlsAuthenticator is certificate-based authentication, using the same certificates used to establish
+#     connection security.
 authenticator:
   class_name: AllowAllAuthenticator
 # MutualTlsAuthenticator can be configured using the following configuration. One can add their own validator
@@ -204,6 +206,31 @@ authenticator:
 #  parameters:
 #    validator_class_name: org.apache.cassandra.auth.SpiffeCertificateValidator
 
+# Configure authenticator negotiation, to allow nodes to support multiple authentication mechanisms and
+# negotiate (resolve) a specific mechanism with a client at connection time. When this configuration is present,
+# it supersedes the 'authenticator' configuration entirely.
+# authenticator_negotiation:
+  # If true, client connections will go through authn negotiation. If false, they will use the authenticator
+  # configured by the default_authenticator key.
+  # enabled: false
+  # If true, all configured authenticators, including the default authenticator, must 'require authenticiation':
+  # i.e., 'AllowAllAuthenticator' and other authenticators that don't require authentication cannot be used and
+  # will cause the node to fail at starttup. This prevents 'fail-open' negotiation. If false, AllowAllAuthenticator
+  # and other non-authenticating authenticators are allowed. This is discouraged but may make migrating to
+  # negotiated authentication easier for deployments that do not enforce authentication today.
+  # require_authentication: true
+  # The authenticator to be used for clients that don't support negotiation, or clients that support none of
+  # the authenticators listed in the 'authenticators' section.
+  # default_authenticator:
+    # class_name: PasswordAuthenticator
+  # The authenticators that this node supports for negotiation. These are configured as documented for the
+  # 'authenticator' section. List authenticators in order from most- to least-preferred. For negotiating
+  # clients, the server will select the first authenticator in this list that the client indicates it can
+  # support.
+  # authenticators:
+    # - class_name: PasswordAuthenticator
+    # - class_name: AllowAllAuthenticator
+
 # Authorization backend, implementing IAuthorizer; used to limit access/provide permissions
 # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllAuthorizer,
 # CassandraAuthorizer}.
@@ -1924,7 +1951,7 @@ client_encryption_options:
   # Custom auth settings which can be used as alternatives to JMX's out of the box auth utilities.
   # JAAS login modules can be used for authentication using this property.Cassandra ships with a
   # LoginModule implementation - org.apache.cassandra.auth.CassandraLoginModule - which delegates
-  # to the IAuthenticator configured in cassandra.yaml.
+  # to the default IAuthenticator configured in cassandra.yaml.
   #
   # login_config_name refers to the Application Name in the JAAS configuration under which the
   # desired LoginModule(s) are configured.

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

@@ -0,0 +1,70 @@
+/*
+ * 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.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+
+/**
+ * Implements the business logic for selecting an authenticator for a given session, when both the node and the
+ * client support authenticator negotiation.
+ */
+public class AuthenticatorNegotiator
+{
+    private static final Logger logger = LoggerFactory.getLogger(AuthenticatorNegotiator.class);
+
+    /**
+     * Selects an {@link IAuthenticator authenticator} based on client-provided authentication modes and this
+     * node's own prioritized list of supported authenticators.
+     *
+     * @param clientAuthenticators Comma-separated list of {@link IAuthenticator.AuthenticationMode} names supported
+     *                             by the client. May be null or empty if the client doesn't support or is not
+     *                             configured for negotiation
+     * @return The node's most preferred authenticator that supports at least one of the client's offered authentication
+     *         modes, or its default authenticator if no modes are provided or none match.
+     */
+    public static IAuthenticator negotiateAuthenticator(@Nonnull Set<String> clientAuthenticators)
+    {
+        Set<IAuthenticator.AuthenticationMode> clientAuthenticationModes =
+            clientAuthenticators.stream()
+                                .map(name -> new IAuthenticator.AuthenticationMode(name) {})
+                                .collect(Collectors.toSet());
+
+        for (IAuthenticator authenticator : DatabaseDescriptor.getNegotiableAuthenticators())
+        {
+            if (!Collections.disjoint(clientAuthenticationModes, authenticator.getSupportedAuthenticationModes()))
+            {
+                logger.info("Negotiated authenticator with client with options {}: selected {}",
+                            clientAuthenticationModes, authenticator.getClass().getName());
+                return authenticator;
+            }
+        }
+
+        logger.info("Auth negotiation failed for client options {}: continuing with default authenticator", clientAuthenticators);
+
+        return DatabaseDescriptor.getDefaultAuthenticator();
+    }
+}

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

@@ -36,11 +36,11 @@
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.exceptions.AuthenticationException;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.CassandraDaemon;
 
 /**
- * LoginModule which authenticates a user towards the Cassandra database using
- * the internal authentication mechanism.
+ * LoginModule which authenticates a user towards the Cassandra database using the default authentication
+ * mechanism. This is used to support JMX authentication.
  */
 public class CassandraLoginModule implements LoginModule
 {
@@ -140,10 +140,10 @@ public boolean login() throws LoginException
 
     private void authenticate()
     {
-        if (!StorageService.instance.isAuthSetupComplete())
+        if (!CassandraDaemon.isAuthSetupComplete())
             throw new AuthenticationException("Cannot login as server authentication setup is not yet completed");
 
-        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
+        IAuthenticator authenticator = DatabaseDescriptor.getDefaultAuthenticator();
         Map<String, String> credentials = new HashMap<>();
         credentials.put(PasswordAuthenticator.USERNAME_KEY, username);
         credentials.put(PasswordAuthenticator.PASSWORD_KEY, String.valueOf(password));

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

@@ -31,6 +31,8 @@
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.transport.messages.AuthenticateMessage;
 
+import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
+
 public interface IAuthenticator
 {
     /**
@@ -248,6 +250,7 @@ default AuthenticationMode getAuthenticationMode()
     abstract class AuthenticationMode
     {
         private final String displayName;
+        private final String normalizedDisplayName;
 
         /**
          * @param displayName How this mode should be displayed in tooling and JMX beans.  Note that it is desirable
@@ -256,6 +259,7 @@ abstract class AuthenticationMode
         public AuthenticationMode(@Nonnull String displayName)
         {
             this.displayName = displayName;
+            this.normalizedDisplayName = toLowerCaseLocalized(displayName);
         }
 
         /**
@@ -283,15 +287,15 @@ public String toString()
         public boolean equals(Object o)
         {
             if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
+            if (!(o instanceof AuthenticationMode)) return false;
             AuthenticationMode that = (AuthenticationMode) o;
-            return displayName.equals(that.displayName);
+            return normalizedDisplayName.equals(that.normalizedDisplayName);
         }
 
         @Override
         public int hashCode()
         {
-            return Objects.hash(displayName);
+            return Objects.hash(normalizedDisplayName);
         }
     }
 }

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

@@ -75,6 +75,7 @@ public class PasswordAuthenticator implements IAuthenticator, AuthCache.BulkLoad
     public static final String USERNAME_KEY = "username";
     public static final String PASSWORD_KEY = "password";
     private static final Set<AuthenticationMode> AUTHENTICATION_MODES = Collections.singleton(AuthenticationMode.PASSWORD);
+    static final byte NUL = 0;
 
     @VisibleForTesting
     static final Set<IRoleManager.Option> SUPPORTED_ROLE_OPTIONS =
@@ -89,17 +90,9 @@ public class PasswordAuthenticator implements IAuthenticator, AuthCache.BulkLoad
                        IRoleManager.Option.HASHED_PASSWORD,
                        IRoleManager.Option.GENERATED_PASSWORD);
 
-    static final byte NUL = 0;
+    private static CredentialsCache cache;
     private SelectStatement authenticateStatement;
 
-    private final CredentialsCache cache;
-
-    public PasswordAuthenticator()
-    {
-        cache = new CredentialsCache(this);
-        AuthCacheService.instance.register(cache);
-    }
-
     /**
      * {@inheritDoc}
      */
@@ -248,6 +241,17 @@ public void validateConfiguration() throws ConfigurationException
 
     public void setup()
     {
+        // Cache initialization is deferred to setup() to avoid duplicate cache registration when subclasses
+        // of PasswordAuthenticator (e.g., MutualTlsWithPasswordFallbackAuthenticator) are also instantiated.
+        synchronized (PasswordAuthenticator.class)
+        {
+            if (cache == null)
+            {
+                cache = new CredentialsCache(this);
+                AuthCacheService.instance.register(cache);
+            }
+        }
+
         String query = String.format("SELECT %s FROM %s.%s WHERE role = ?",
                                      SALTED_HASH,
                                      SchemaConstants.AUTH_KEYSPACE_NAME,

src/java/org/apache/cassandra/auth/jmx/AuthorizationProxy.java

@@ -53,7 +53,7 @@
 import org.apache.cassandra.auth.RoleResource;
 import org.apache.cassandra.auth.Roles;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.CassandraDaemon;
 import org.apache.cassandra.utils.JmxInvocationListener;
 import org.apache.cassandra.utils.MBeanWrapper;
 
@@ -148,10 +148,10 @@ public class AuthorizationProxy implements InvocationHandler
     protected Function<ObjectName, Set<ObjectName>> queryNames = (name) -> mbs.queryNames(name, null);
 
     /*
-     Used to determine whether auth setup has completed so we know whether the expect the IAuthorizer
+     Used to determine whether auth setup has completed so we know whether to expect the IAuthorizer
      to be ready. Can be overridden for testing.
      */
-    protected BooleanSupplier isAuthSetupComplete = () -> StorageService.instance.isAuthSetupComplete();
+    protected BooleanSupplier isAuthSetupComplete = () -> CassandraDaemon.isAuthSetupComplete();
 
     protected JmxInvocationListener listener = AuditLogManager.instance;
 

src/java/org/apache/cassandra/config/Config.java

@@ -19,10 +19,12 @@
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
@@ -83,6 +85,23 @@ public static Set<String> splitCommaDelimited(String src)
     public static final String PROPERTY_PREFIX = "cassandra.";
 
     public String cluster_name = "Test Cluster";
+
+    /**
+     * Configuration for authenticator negotiation, enabling nodes to support multiple authentication mechanisms
+     * and negotiate with clients at connection time.
+     * <p/>
+     * When configured, these settings take precedence over the legacy 'authenticator' configuration. The
+     * 'default_authenticator' is used for non-negotiating clients or when negotiation fails to resolve.
+     */
+    public static class AuthenticatorNegotiationConfig
+    {
+        public boolean enabled = false;
+        public boolean require_authentication = true;
+        public ParameterizedClass default_authenticator;
+        public List<ParameterizedClass> authenticators = new ArrayList<>();
+    }
+
+    public AuthenticatorNegotiationConfig authenticator_negotiation = new AuthenticatorNegotiationConfig();
     public ParameterizedClass authenticator;
     public ParameterizedClass authorizer;
     public ParameterizedClass role_manager;