From 281c1dc92ccb09c57ed0eeccc3d7935b4e051ebc Mon Sep 17 00:00:00 2001
From: Joel Shepherd <shepherd@amazon.com>
Date: Tue, 14 Oct 2025 00:12:10 +0000
Subject: [PATCH] Enable IAuthenticator to declare supported and alterable role
 options

With negotiated authentication (CEP-50), nodes may be configured with multiple authenticators. Prior to this change,
a number of areas in the code assumed that there was a single configured authenticator and contained logic that
switched depending on the authenticator type. This logic won't work when multiple authenticators can be configured.
This change eliminates most calls to DataDescriptor.getAuthenticator(), by enabling individual authenticators to
declare the role attributes they support, requiring callers to specify the type of authenticator they're looking
for, and directly returning whether the node can enforce authn or not rather than inferring it by the presence of
an authenticator.

Testing done: Unit tests for auth and config packages; d-tests for auth-related functionality (e.g. ColumnMasks).

patch by Joel Shepherd; reviewed by Stefan Miklosovic, Andy Tolbert for CASSANDRA-20834
---
 CHANGES.txt                                   |   1 +
 .../org/apache/cassandra/auth/AuthConfig.java |  12 ++
 .../cassandra/auth/CassandraRoleManager.java  |  72 ++++++-----
 .../apache/cassandra/auth/IAuthenticator.java |  23 ++++
 .../auth/MutualTlsAuthenticator.java          |   5 +-
 .../auth/NetworkPermissionsCache.java         |   2 +-
 .../cassandra/auth/PasswordAuthenticator.java |  37 +++++-
 src/java/org/apache/cassandra/auth/Roles.java |   2 +-
 .../cassandra/config/DatabaseDescriptor.java  |  36 ++++++
 .../db/virtual/CredentialsCacheKeysTable.java |  15 +--
 .../cassandra/service/CassandraDaemon.java    |   2 +-
 .../apache/cassandra/service/ClientState.java |   4 +-
 test/conf/cassandra-passwordauth.yaml         |  93 ++++++++++++++
 .../apache/cassandra/auth/AuthConfigTest.java | 117 ++++++++++++++----
 14 files changed, 347 insertions(+), 74 deletions(-)
 create mode 100644 test/conf/cassandra-passwordauth.yaml

diff --git a/CHANGES.txt b/CHANGES.txt
index bceda5b9d8..579b190f00 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 6.0-alpha2
+ * Enable IAuthenticator to declare supported and alterable role options (CASSANDRA-20834)
  * Avoid capturing lambda allocation in UnfilteredSerializer.deserializeRowBody (CASSANDRA-21289)
  * Avoid Cell iterator allocation for alive rows in MetricsRecording transformation of ReadCommand (CASSANDRA-21288)
  * Reuse a single TrackedDataInputPlus instance per UnfilteredSerializer (CASSANDRA-21296)
diff --git a/src/java/org/apache/cassandra/auth/AuthConfig.java b/src/java/org/apache/cassandra/auth/AuthConfig.java
index 67d4490790..86e1f626ff 100644
--- a/src/java/org/apache/cassandra/auth/AuthConfig.java
+++ b/src/java/org/apache/cassandra/auth/AuthConfig.java
@@ -20,6 +20,8 @@ package org.apache.cassandra.auth;
 
 import java.util.List;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,6 +40,16 @@ public final class AuthConfig
 
     private static boolean initialized;
 
+    /**
+     * Resets the initialized flag, enabling AuthConfig to be reconfigured multiple times within a single
+     * test case.
+     */
+    @VisibleForTesting
+    static void reset()
+    {
+        initialized = false;
+    }
+
     public static void applyAuth()
     {
         // some tests need this
diff --git a/src/java/org/apache/cassandra/auth/CassandraRoleManager.java b/src/java/org/apache/cassandra/auth/CassandraRoleManager.java
index 791ff02ba1..ec4f873b5c 100644
--- a/src/java/org/apache/cassandra/auth/CassandraRoleManager.java
+++ b/src/java/org/apache/cassandra/auth/CassandraRoleManager.java
@@ -21,7 +21,6 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -91,29 +90,16 @@ import static org.apache.cassandra.service.QueryState.forInternalCalls;
 import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
 
 /**
- * Responsible for the creation, maintenance and deletion of roles
- * for the purposes of authentication and authorization.
- * Role data is stored internally, using the roles and role_members tables
- * in the system_auth keyspace.
+ * Responsible for the creation, maintenance and deletion of roles for the purposes of authentication and
+ * authorization. Role data is stored internally, using the roles and role_members tables in the system_auth
+ * keyspace.
  *
- * Additionally, if org.apache.cassandra.auth.PasswordAuthenticator is used,
- * encrypted passwords are also stored in the system_auth.roles table. This
- * coupling between the IAuthenticator and IRoleManager implementations exists
- * because setting a role's password via CQL is done with a CREATE ROLE or
- * ALTER ROLE statement, the processing of which is handled by IRoleManager.
- * As IAuthenticator is concerned only with credentials checking and has no
- * means to modify passwords, PasswordAuthenticator depends on
- * CassandraRoleManager for those functions.
- *
- * Alternative IAuthenticator implementations may be used in conjunction with
- * CassandraRoleManager, but WITH PASSWORD = 'password' will not be supported
- * in CREATE/ALTER ROLE statements.
- *
- * Such a configuration could be implemented using a custom IRoleManager that
- * extends CassandraRoleManager and which includes Option.PASSWORD in the {@code Set<Option>}
- * returned from supportedOptions/alterableOptions. Any additional processing
- * of the password itself (such as storing it in an alternative location) would
- * be added in overridden createRole and alterRole implementations.
+ * Authenticators (implementations of {@link IAuthenticator}) can specify additional attributes to be stored.
+ * For example, {@link org.apache.cassandra.auth.PasswordAuthenticator}, stores encrypted passwords in the
+ * system_auth.roles table. This coupling between the IAuthenticator and IRoleManager implementations exists because
+ * setting a role's password via CQL is done with a CREATE ROLE or ALTER ROLE statement, the processing of which is
+ * handled by IRoleManager. Authenticators depend on CassandraRoleManager for those functions because IAuthenticator
+ * is concerned only with credentials checking and has no means to directly modify passwords.
  */
 public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerMBean
 {
@@ -123,8 +109,24 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
     public static final String DEFAULT_SUPERUSER_NAME = "cassandra";
     public static final String DEFAULT_SUPERUSER_PASSWORD = "cassandra";
 
+    /**
+     * Role options which are supported for all authentication mechanisms. IAuthenticator implementations can declare
+     * additional supported role options via {@link IAuthenticator#getSupportedRoleOptions()}.
+     */
+    @VisibleForTesting
+    static final Set<Option> DEFAULT_SUPPORTED_ROLE_OPTIONS = Set.of(Option.LOGIN, Option.SUPERUSER);
+
+    /**
+     * User-alterable role options which are supported for all authentication mechanisms. IAuthenticator
+     * implementations can declare additional alterable role options via
+     * {@link IAuthenticator#getAlterableRoleOptions()}.
+     */
+    @VisibleForTesting
+    static final Set<Option> DEFAULT_ALTERABLE_ROLE_OPTIONS = Set.of();
+
     @VisibleForTesting
     static final String PARAM_INVALID_ROLE_DISCONNECT_TASK_PERIOD = "invalid_role_disconnect_task_period";
+
     @VisibleForTesting
     static final String PARAM_INVALID_ROLE_DISCONNECT_TASK_MAX_JITTER = "invalid_role_disconnect_task_max_jitter";
 
@@ -191,17 +193,21 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
 
     public CassandraRoleManager(Map<String, String> parameters)
     {
-        Set<Option> allowedOptions = DatabaseDescriptor.getAuthenticator() instanceof PasswordAuthenticator
-                                     ? EnumSet.of(Option.LOGIN, Option.SUPERUSER, Option.PASSWORD, Option.HASHED_PASSWORD, Option.GENERATED_PASSWORD, Option.GENERATED_NAME)
-                                     : EnumSet.of(Option.LOGIN, Option.SUPERUSER);
+        Set<Option> supportedOptions = Stream.concat(
+                                           DEFAULT_SUPPORTED_ROLE_OPTIONS.stream(),
+                                           DatabaseDescriptor.getAuthenticator().getSupportedRoleOptions().stream()
+                                                             .filter(Objects::nonNull))
+                                             .collect(Collectors.toSet());
 
         if (Guardrails.roleNamePolicy.getGenerator() != NoOpGenerator.INSTANCE)
-            allowedOptions.add(Option.OPTIONS);
+            supportedOptions.add(Option.OPTIONS);
 
-        supportedOptions = ImmutableSet.copyOf(allowedOptions);
-        alterableOptions = DatabaseDescriptor.getAuthenticator() instanceof PasswordAuthenticator
-                           ? ImmutableSet.of(Option.PASSWORD, Option.HASHED_PASSWORD, Option.GENERATED_PASSWORD)
-                           : ImmutableSet.<Option>of();
+        this.supportedOptions = Set.copyOf(supportedOptions);
+
+        alterableOptions = Stream.concat(DEFAULT_ALTERABLE_ROLE_OPTIONS.stream(),
+                                         DatabaseDescriptor.getAuthenticator().getAlterableRoleOptions().stream()
+                                                           .filter(Objects::nonNull))
+                                 .collect(Collectors.toUnmodifiableSet());
 
         // Inherit parsing and validation from existing config parser
         invalidClientDisconnectPeriodMillis = new DurationSpec.LongMillisecondsBound(parameters.getOrDefault(PARAM_INVALID_ROLE_DISCONNECT_TASK_PERIOD, "0h")).toMilliseconds();
@@ -505,8 +511,8 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
 
     public Set<? extends IResource> protectedResources()
     {
-        return ImmutableSet.of(DataResource.table(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES),
-                               DataResource.table(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLE_MEMBERS));
+        return Set.of(DataResource.table(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES),
+                      DataResource.table(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLE_MEMBERS));
     }
 
     public void validateConfiguration() throws ConfigurationException
diff --git a/src/java/org/apache/cassandra/auth/IAuthenticator.java b/src/java/org/apache/cassandra/auth/IAuthenticator.java
index e58c6ab2bb..a4e888e505 100644
--- a/src/java/org/apache/cassandra/auth/IAuthenticator.java
+++ b/src/java/org/apache/cassandra/auth/IAuthenticator.java
@@ -66,6 +66,29 @@ public interface IAuthenticator
      */
     Set<? extends IResource> protectedResources();
 
+    /**
+     * Additional set of IRoleManager.Options used by this authenticator and supported by CREATE ROLE and ALTER ROLE
+     * statements. These are in addition to any default options supported by the role manager.
+     *
+     * @return A set of IRoleManager.Options that this authenticator requires support for.
+     */
+    default Set<IRoleManager.Option> getSupportedRoleOptions()
+    {
+        return Set.of();
+    }
+
+    /**
+     * Additional set of IRoleManager.Options used by this authenticator that users are allowed to alter via
+     * ALTER ROLE statements. These are in addition to any default alterable options supported by the role manager.
+     * Alterable role options must also be supported role options.
+     *
+     * @return A set of supported role options that users are allowed to alter.
+     */
+    default Set<IRoleManager.Option> getAlterableRoleOptions()
+    {
+        return Set.of();
+    }
+
     /**
      * Validates configuration of IAuthenticator implementation (if configurable).
      *
diff --git a/src/java/org/apache/cassandra/auth/MutualTlsAuthenticator.java b/src/java/org/apache/cassandra/auth/MutualTlsAuthenticator.java
index e80e3156b1..76aacc5b5a 100644
--- a/src/java/org/apache/cassandra/auth/MutualTlsAuthenticator.java
+++ b/src/java/org/apache/cassandra/auth/MutualTlsAuthenticator.java
@@ -78,13 +78,16 @@ public class MutualTlsAuthenticator implements IAuthenticator
     private static final Logger logger = LoggerFactory.getLogger(MutualTlsAuthenticator.class);
     private static final NoSpamLogger nospamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES);
     private static final String VALIDATOR_CLASS_NAME = "validator_class_name";
-    private static final String CACHE_NAME = "IdentitiesCache";
+
     private final IdentityCache identityCache = new IdentityCache();
     private final MutualTlsCertificateValidator certificateValidator;
     private static final Set<AuthenticationMode> AUTHENTICATION_MODES = Collections.singleton(MTLS);
     private final MutualTlsCertificateValidityPeriodValidator certificateValidityPeriodValidator;
     private final DurationSpec.IntMinutesBound certificateValidityWarnThreshold;
 
+    @VisibleForTesting
+    static final String CACHE_NAME = "IdentitiesCache";
+
     // key for the 'identity' value in AuthenticatedUser metadata map.
     public static final String METADATA_IDENTITY_KEY = "identity";
 
diff --git a/src/java/org/apache/cassandra/auth/NetworkPermissionsCache.java b/src/java/org/apache/cassandra/auth/NetworkPermissionsCache.java
index 1c18fed7a7..f232cf354f 100644
--- a/src/java/org/apache/cassandra/auth/NetworkPermissionsCache.java
+++ b/src/java/org/apache/cassandra/auth/NetworkPermissionsCache.java
@@ -36,7 +36,7 @@ public class NetworkPermissionsCache extends AuthCache<RoleResource, DCPermissio
               DatabaseDescriptor::getRolesCacheActiveUpdate,
               authorizer::authorize,
               authorizer.bulkLoader(),
-              () -> DatabaseDescriptor.getAuthenticator().requireAuthentication());
+              DatabaseDescriptor::isAuthenticationRequired);
 
         MBeanWrapper.instance.registerMBean(this, MBEAN_NAME_BASE + DEPRECATED_CACHE_NAME);
     }
diff --git a/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java b/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java
index 82c8f24c6f..9b810f644b 100644
--- a/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java
+++ b/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java
@@ -21,13 +21,13 @@ import java.net.InetAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
 
 import org.mindrot.jbcrypt.BCrypt;
@@ -76,6 +76,19 @@ public class PasswordAuthenticator implements IAuthenticator, AuthCache.BulkLoad
     public static final String PASSWORD_KEY = "password";
     private static final Set<AuthenticationMode> AUTHENTICATION_MODES = Collections.singleton(AuthenticationMode.PASSWORD);
 
+    @VisibleForTesting
+    static final Set<IRoleManager.Option> SUPPORTED_ROLE_OPTIONS =
+            EnumSet.of(IRoleManager.Option.PASSWORD,
+                       IRoleManager.Option.HASHED_PASSWORD,
+                       IRoleManager.Option.GENERATED_PASSWORD,
+                       IRoleManager.Option.GENERATED_NAME);
+
+    @VisibleForTesting
+    static final Set<IRoleManager.Option> ALTERABLE_ROLE_OPTIONS =
+            EnumSet.of(IRoleManager.Option.PASSWORD,
+                       IRoleManager.Option.HASHED_PASSWORD,
+                       IRoleManager.Option.GENERATED_PASSWORD);
+
     static final byte NUL = 0;
     private SelectStatement authenticateStatement;
 
@@ -87,7 +100,26 @@ public class PasswordAuthenticator implements IAuthenticator, AuthCache.BulkLoad
         AuthCacheService.instance.register(cache);
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Set<IRoleManager.Option> getSupportedRoleOptions()
+    {
+        return SUPPORTED_ROLE_OPTIONS;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Set<IRoleManager.Option> getAlterableRoleOptions()
+    {
+        return ALTERABLE_ROLE_OPTIONS;
+    }
+
     // No anonymous access.
+    @Override
     public boolean requireAuthentication()
     {
         return true;
@@ -207,7 +239,7 @@ public class PasswordAuthenticator implements IAuthenticator, AuthCache.BulkLoad
     public Set<DataResource> protectedResources()
     {
         // Also protected by CassandraRoleManager, but the duplication doesn't hurt and is more explicit
-        return ImmutableSet.of(DataResource.table(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES));
+        return Set.of(DataResource.table(SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES));
     }
 
     public void validateConfiguration() throws ConfigurationException
@@ -347,6 +379,7 @@ public class PasswordAuthenticator implements IAuthenticator, AuthCache.BulkLoad
                                                      // invalidate the key if the sentinel is loaded during a refresh
         }
 
+        @Override
         public void invalidateCredentials(String roleName)
         {
             invalidate(roleName);
diff --git a/src/java/org/apache/cassandra/auth/Roles.java b/src/java/org/apache/cassandra/auth/Roles.java
index c4070aaff3..1c5ec0b008 100644
--- a/src/java/org/apache/cassandra/auth/Roles.java
+++ b/src/java/org/apache/cassandra/auth/Roles.java
@@ -36,7 +36,7 @@ public class Roles
 
     private static final Role NO_ROLE = new Role("", false, false, Collections.emptyMap(), Collections.emptySet());
 
-    public static final RolesCache cache = new RolesCache(DatabaseDescriptor.getRoleManager(), () -> DatabaseDescriptor.getAuthenticator().requireAuthentication());
+    public static final RolesCache cache = new RolesCache(DatabaseDescriptor.getRoleManager(), DatabaseDescriptor::isAuthenticationRequired);
 
     /** Use {@link AuthCacheService#initializeAndRegisterCaches} rather than calling this directly */
     public static void init()
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 59ffb9f02a..b8bd66bdf5 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -2037,16 +2037,52 @@ public class DatabaseDescriptor
         DatabaseDescriptor.cryptoProvider = cryptoProvider;
     }
 
+    /**
+     * Returns the authenticator configured for this node.
+     */
     public static IAuthenticator getAuthenticator()
     {
         return authenticator;
     }
 
+    /**
+     * Returns an authenticator configured for this node, if it is of the requested type.
+     * @param clazz The class of the requested authenticator: e.g. PasswordAuthenticator.class.
+     * @return An Optional of the configured authenticator, if it is of the requested type; otherwise
+     *         returns an empty Optional.
+     */
+    public static <T extends IAuthenticator> Optional<T> getAuthenticator(Class<T> clazz)
+    {
+        return hasAuthenticator(clazz) ? Optional.of(clazz.cast(authenticator)) : Optional.empty();
+    }
+
+    /**
+     * Sets the authenticator used by this node to authenticate clients.
+     */
     public static void setAuthenticator(IAuthenticator authenticator)
     {
         DatabaseDescriptor.authenticator = authenticator;
     }
 
+    /**
+     * Indicates if this node uses an authenticator that requires authentication.
+     */
+    public static boolean isAuthenticationRequired()
+    {
+        return authenticator.requireAuthentication();
+    }
+
+    /**
+     * Indicates if this node is configured with an authenticator of the specified type.
+     * @param clazz The class of the authenticator.
+     * @return True if this node has an authenticator of the specified type, false otherwise.
+     */
+    private static boolean hasAuthenticator(Class<? extends IAuthenticator> clazz)
+    {
+        return clazz.isAssignableFrom(authenticator.getClass());
+    }
+
+
     public static IAuthorizer getAuthorizer()
     {
         return authorizer;
diff --git a/src/java/org/apache/cassandra/db/virtual/CredentialsCacheKeysTable.java b/src/java/org/apache/cassandra/db/virtual/CredentialsCacheKeysTable.java
index f5bc62c526..0040f02547 100644
--- a/src/java/org/apache/cassandra/db/virtual/CredentialsCacheKeysTable.java
+++ b/src/java/org/apache/cassandra/db/virtual/CredentialsCacheKeysTable.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.db.virtual;
 
 import java.util.Optional;
 
-import org.apache.cassandra.auth.IAuthenticator;
 import org.apache.cassandra.auth.PasswordAuthenticator;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.marshal.UTF8Type;
@@ -31,7 +30,7 @@ final class CredentialsCacheKeysTable extends AbstractMutableVirtualTable
     private static final String ROLE = "role";
 
     @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
-    private final Optional<PasswordAuthenticator> passwordAuthenticatorOptional;
+    private final Optional<PasswordAuthenticator> maybePasswordAuthenticator;
 
     CredentialsCacheKeysTable(String keyspace)
     {
@@ -42,18 +41,14 @@ final class CredentialsCacheKeysTable extends AbstractMutableVirtualTable
                 .addPartitionKeyColumn(ROLE, UTF8Type.instance)
                 .build());
 
-        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
-        if (authenticator instanceof PasswordAuthenticator)
-            this.passwordAuthenticatorOptional = Optional.of((PasswordAuthenticator) authenticator);
-        else
-            this.passwordAuthenticatorOptional = Optional.empty();
+        maybePasswordAuthenticator = DatabaseDescriptor.getAuthenticator(PasswordAuthenticator.class);
     }
 
     public DataSet data()
     {
         SimpleDataSet result = new SimpleDataSet(metadata());
 
-        passwordAuthenticatorOptional
+        maybePasswordAuthenticator
                 .ifPresent(passwordAuthenticator -> passwordAuthenticator.getCredentialsCache().getAll()
                         .forEach((roleName, ignored) -> result.row(roleName)));
 
@@ -65,14 +60,14 @@ final class CredentialsCacheKeysTable extends AbstractMutableVirtualTable
     {
         String roleName = partitionKey.value(0);
 
-        passwordAuthenticatorOptional
+        maybePasswordAuthenticator
                 .ifPresent(passwordAuthenticator -> passwordAuthenticator.getCredentialsCache().invalidate(roleName));
     }
 
     @Override
     public void truncate()
     {
-        passwordAuthenticatorOptional
+        maybePasswordAuthenticator
                 .ifPresent(passwordAuthenticator -> passwordAuthenticator.getCredentialsCache().invalidate());
     }
 }
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 4a5c2f1d9d..8c8c53ad3e 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -901,7 +901,7 @@ public class CassandraDaemon
         {
             if (StorageService.instance.isSurveyMode())
             {
-                if (!StorageService.instance.readyToFinishJoiningRing() || DatabaseDescriptor.getAuthenticator().requireAuthentication())
+                if (!StorageService.instance.readyToFinishJoiningRing() || DatabaseDescriptor.isAuthenticationRequired())
                 {
                     throw new IllegalStateException("Not starting client transports in write_survey mode as it's bootstrapping or " +
                                                     "auth is enabled");
diff --git a/src/java/org/apache/cassandra/service/ClientState.java b/src/java/org/apache/cassandra/service/ClientState.java
index 7ec2160e8d..c119d5f4c3 100644
--- a/src/java/org/apache/cassandra/service/ClientState.java
+++ b/src/java/org/apache/cassandra/service/ClientState.java
@@ -199,7 +199,7 @@ public class ClientState
     {
         this.isInternal = false;
         this.remoteAddress = remoteAddress;
-        if (!DatabaseDescriptor.getAuthenticator().requireAuthentication())
+        if (!DatabaseDescriptor.isAuthenticationRequired())
             this.user = AuthenticatedUser.ANONYMOUS_USER;
     }
 
@@ -628,7 +628,7 @@ public class ClientState
      */
     public boolean isSuper()
     {
-        return !DatabaseDescriptor.getAuthenticator().requireAuthentication() || (user != null && user.isSuper());
+        return !DatabaseDescriptor.isAuthenticationRequired() || (user != null && user.isSuper());
     }
 
     /**
diff --git a/test/conf/cassandra-passwordauth.yaml b/test/conf/cassandra-passwordauth.yaml
new file mode 100644
index 0000000000..42adbb2c6a
--- /dev/null
+++ b/test/conf/cassandra-passwordauth.yaml
@@ -0,0 +1,93 @@
+#
+# 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.
+#
+
+#
+# Minimal cassandra.yaml for testing PasswordAuthenticator integration,
+# particularly with RoleManager.
+#
+cluster_name: Test Cluster
+memtable_allocation_type: offheap_objects
+commitlog_sync: periodic
+commitlog_sync_period: 10s
+commitlog_segment_size: 5MiB
+commitlog_directory: build/test/cassandra/commitlog
+cdc_raw_directory: build/test/cassandra/cdc_raw
+cdc_enabled: false
+hints_directory: build/test/cassandra/hints
+partitioner: org.apache.cassandra.dht.ByteOrderedPartitioner
+listen_address: 127.0.0.1
+storage_port: 7012
+ssl_storage_port: 17012
+start_native_transport: true
+native_transport_port: 9042
+column_index_size: 4KiB
+saved_caches_directory: build/test/cassandra/saved_caches
+data_file_directories:
+  - build/test/cassandra/data
+disk_access_mode: mmap_index_only
+seed_provider:
+  - class_name: org.apache.cassandra.locator.SimpleSeedProvider
+    parameters:
+      - seeds: "127.0.0.1:7012"
+endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch
+dynamic_snitch: true
+incremental_backups: true
+concurrent_compactors: 4
+compaction_throughput: 0MiB/s
+row_cache_class_name: org.apache.cassandra.cache.OHCProvider
+row_cache_size: 16MiB
+prepared_statements_cache_size: 1MiB
+corrupted_tombstone_strategy: exception
+stream_entire_sstables: true
+stream_throughput_outbound: 23841858MiB/s
+sasi_indexes_enabled: true
+materialized_views_enabled: true
+drop_compact_storage_enabled: true
+file_cache_enabled: true
+auto_hints_cleanup_enabled: true
+default_keyspace_rf: 1
+
+client_encryption_options:
+  enabled: true
+  require_client_auth: true
+  keystore: test/conf/cassandra_ssl_test.keystore
+  keystore_password: cassandra
+  truststore: test/conf/cassandra_ssl_test.truststore
+  truststore_password: cassandra
+
+server_encryption_options:
+  internode_encryption: all
+  enabled: true
+  keystore: test/conf/cassandra_ssl_test.keystore
+  keystore_password: cassandra
+  outbound_keystore: test/conf/cassandra_ssl_test_outbound.keystore
+  outbound_keystore_password: cassandra
+  truststore: test/conf/cassandra_ssl_test.truststore
+  truststore_password: cassandra
+  require_client_auth: true
+internode_authenticator:
+  class_name : org.apache.cassandra.auth.MutualTlsInternodeAuthenticator
+  parameters :
+    validator_class_name: org.apache.cassandra.auth.SpiffeCertificateValidator
+authenticator:
+  class_name : org.apache.cassandra.auth.PasswordAuthenticator
+role_manager:
+  class_name: CassandraRoleManager
+  parameters:
+accord:
+  journal_directory: build/test/cassandra/accord_journal
diff --git a/test/unit/org/apache/cassandra/auth/AuthConfigTest.java b/test/unit/org/apache/cassandra/auth/AuthConfigTest.java
index 580c48eb9f..ecbc72f0d5 100644
--- a/test/unit/org/apache/cassandra/auth/AuthConfigTest.java
+++ b/test/unit/org/apache/cassandra/auth/AuthConfigTest.java
@@ -24,28 +24,46 @@ import java.security.cert.CertificateException;
 import java.util.Arrays;
 import java.util.Collections;
 
-import org.junit.Rule;
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
-import org.junit.rules.ExpectedException;
 
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.Builder;
 import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.utils.MBeanWrapper;
 
+import static org.apache.cassandra.auth.AuthCache.MBEAN_NAME_BASE;
 import static org.apache.cassandra.auth.AuthTestUtils.loadCertificateChain;
 import static org.apache.cassandra.auth.IInternodeAuthenticator.InternodeConnectionDirection.INBOUND;
 import static org.apache.cassandra.config.YamlConfigurationLoaderTest.load;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
+/**
+ * Tests instantiation of various authenticators through AuthConfig, and the accessibility of the configured
+ * authenticators and role manager options through DatabaseDescriptor.
+ */
 public class AuthConfigTest
 {
-    @Rule
-    public ExpectedException expectedException = ExpectedException.none();
+    private static final String IDENTITIES_CACHE_MBEAN = MBEAN_NAME_BASE + MutualTlsAuthenticator.CACHE_NAME;
+
+    private static final String CREDENTIALS_CACHE_MBEAN = MBEAN_NAME_BASE + PasswordAuthenticator.CredentialsCacheMBean.CACHE_NAME;
+
+    @Before
+    public void setup()
+    {
+        AuthConfig.reset();
+    }
+
+    @After
+    public void teardown()
+    {
+        unregisterCaches();
+    }
 
     @Test
     public void testNewInstanceForMutualTlsInternodeAuthenticator() throws IOException, CertificateException
@@ -53,13 +71,11 @@ public class AuthConfigTest
         Config config = load("cassandra-mtls.yaml");
         config.internode_authenticator.class_name = "org.apache.cassandra.auth.MutualTlsInternodeAuthenticator";
         config.internode_authenticator.parameters = Collections.singletonMap("validator_class_name", "org.apache.cassandra.auth.SpiffeCertificateValidator");
-        config.server_encryption_options = new Builder(config.server_encryption_options)
-                                           .withOutboundKeystore("test/conf/cassandra_ssl_test_outbound.keystore")
-                                           .withOutboundKeystorePassword("cassandra")
-                                           .build();
-        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.unsafeDaemonInitialization(()->config);
+
         MutualTlsInternodeAuthenticator authenticator = ParameterizedClass.newInstance(config.internode_authenticator,
                                                                                        Arrays.asList("", "org.apache.cassandra.auth."));
+        assertNotNull(authenticator);
 
         InetAddressAndPort address = InetAddressAndPort.getByName("127.0.0.1");
 
@@ -68,36 +84,91 @@ public class AuthConfigTest
 
         Certificate[] unauthorizedCertificates = loadCertificateChain("auth/SampleUnauthorizedMtlsClientCertificate.pem");
         assertFalse(authenticator.authenticate(address.getAddress(), address.getPort(), unauthorizedCertificates, INBOUND));
+        unregisterCaches();
+    }
+
+    @Test
+    public void testNewInstanceForPasswordAuthenticator()
+    {
+        Config config = load("cassandra-passwordauth.yaml");
+        DatabaseDescriptor.unsafeDaemonInitialization(()->config);
+
+        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
+        assertNotNull(authenticator);
+
+        assertThat(DatabaseDescriptor.getAuthenticator(PasswordAuthenticator.class))
+            .isPresent()
+            .get()
+            .isSameAs(authenticator);
+        assertThat(DatabaseDescriptor.getAuthenticator(MutualTlsAuthenticator.class)).isEmpty();
+        assertThat(DatabaseDescriptor.getAuthenticator(MutualTlsWithPasswordFallbackAuthenticator.class)).isEmpty();
+        assertTrue(DatabaseDescriptor.getRoleManager().supportedOptions().containsAll(CassandraRoleManager.DEFAULT_SUPPORTED_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().supportedOptions().containsAll(PasswordAuthenticator.SUPPORTED_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(CassandraRoleManager.DEFAULT_ALTERABLE_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(PasswordAuthenticator.ALTERABLE_ROLE_OPTIONS));
     }
 
     @Test
     public void testNewInstanceForMutualTlsWithPasswordFallbackAuthenticator()
     {
         Config config = load("cassandra-mtls.yaml");
-        config.client_encryption_options.applyConfig();
         config.authenticator.class_name = "org.apache.cassandra.auth.MutualTlsWithPasswordFallbackAuthenticator";
         config.authenticator.parameters = Collections.singletonMap("validator_class_name", "org.apache.cassandra.auth.SpiffeCertificateValidator");
-        DatabaseDescriptor.setConfig(config);
-        MutualTlsWithPasswordFallbackAuthenticator authenticator = ParameterizedClass.newInstance(config.authenticator,
-                                                                                                  Arrays.asList("", "org.apache.cassandra.auth."));
+        DatabaseDescriptor.unsafeDaemonInitialization(()->config);
+
+        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
         assertNotNull(authenticator);
-        unregisterIdentitesCache();
+
+        // MutualTlsWithPasswordFallbackAuthenticator is-a PasswordAuthenticator, so we expect getAuthenticator to
+        // return it when asked to return a PasswordAuthenticator.
+        assertThat(DatabaseDescriptor.getAuthenticator(PasswordAuthenticator.class))
+            .isPresent()
+            .get()
+            .isSameAs(authenticator);
+        assertThat(DatabaseDescriptor.getAuthenticator(MutualTlsAuthenticator.class)).isEmpty();
+        assertThat(DatabaseDescriptor.getAuthenticator(MutualTlsWithPasswordFallbackAuthenticator.class))
+            .isPresent()
+            .get()
+            .isSameAs(authenticator);
+
+        // Similarly, we expect the same role options as for PasswordAuthenticator.
+        assertTrue(DatabaseDescriptor.getRoleManager().supportedOptions().containsAll(CassandraRoleManager.DEFAULT_SUPPORTED_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().supportedOptions().containsAll(PasswordAuthenticator.SUPPORTED_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(CassandraRoleManager.DEFAULT_ALTERABLE_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(PasswordAuthenticator.ALTERABLE_ROLE_OPTIONS));
     }
 
     @Test
-    public void testNewInstanceForMutualTlsAuthenticator() throws IOException, CertificateException
+    public void testNewInstanceForMutualTlsAuthenticator()
     {
         Config config = load("cassandra-mtls.yaml");
-        config.client_encryption_options.applyConfig();
-        DatabaseDescriptor.setConfig(config);
-        MutualTlsAuthenticator authenticator = ParameterizedClass.newInstance(config.authenticator,
-                                                                              Arrays.asList("", "org.apache.cassandra.auth."));
+        DatabaseDescriptor.unsafeDaemonInitialization(()->config);
+
+        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
         assertNotNull(authenticator);
-        unregisterIdentitesCache();
+
+        assertThat(DatabaseDescriptor.getAuthenticator(PasswordAuthenticator.class)).isEmpty();
+        assertThat(DatabaseDescriptor.getAuthenticator(MutualTlsAuthenticator.class))
+            .isPresent()
+            .get()
+            .isSameAs(authenticator);
+        assertThat(DatabaseDescriptor.getAuthenticator(MutualTlsWithPasswordFallbackAuthenticator.class)).isEmpty();
+
+        assertTrue(DatabaseDescriptor.getRoleManager().supportedOptions().containsAll(CassandraRoleManager.DEFAULT_SUPPORTED_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().supportedOptions().containsAll(authenticator.getSupportedRoleOptions()));
+        assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(CassandraRoleManager.DEFAULT_ALTERABLE_ROLE_OPTIONS));
+        assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(authenticator.getAlterableRoleOptions()));
     }
 
-    private void unregisterIdentitesCache()
+    private void unregisterCaches()
     {
-        MBeanWrapper.instance.unregisterMBean("org.apache.cassandra.auth:type=IdentitiesCache");
+        safeUnregisterMbean(IDENTITIES_CACHE_MBEAN);
+        safeUnregisterMbean(CREDENTIALS_CACHE_MBEAN);
+    }
+
+    private void safeUnregisterMbean(String mbeanName)
+    {
+        if (MBeanWrapper.instance.isRegistered(mbeanName))
+            MBeanWrapper.instance.unregisterMBean(mbeanName);
     }
 }