Merge 96a5f44eed into 10557d7ffe

conf/cassandra.yaml

@@ -241,6 +241,30 @@ role_manager:
     # invalid_role_disconnect_task_period: 4h
     # invalid_role_disconnect_task_max_jitter: 1h
 
+# Creates the initial role on a cluster which has no roles yet, implementing IDefaultRoleInitializer.
+# Most functions of the IRoleManager require an authenticated login, so a cluster with no roles has no way
+# to create the first one; this option controls how that role is bootstrapped.
+#
+# Defaults to PasswordDefaultRoleInitializer, which creates a 'cassandra' superuser whose password is also
+# 'cassandra'. That password is a published constant, so deployments using it must rotate or drop the role
+# before the native transport is reachable.
+#
+# MutualTlsDefaultRoleInitializer instead creates the role with no password at all and maps a client
+# certificate identity onto it, so there is no credential to guess. It requires an authenticator supporting
+# mutual TLS, such as MutualTlsAuthenticator.
+#
+# default_role_initializer:
+#   class_name: PasswordDefaultRoleInitializer
+#   parameters:
+#     role: cassandra
+#     password: cassandra
+#
+# default_role_initializer:
+#   class_name: MutualTlsDefaultRoleInitializer
+#   parameters:
+#     role: cassandra_mtls
+#     identity: "spiffe1"
+
 # Network authorization backend, implementing INetworkAuthorizer; used to restrict user
 # access to certain DCs
 # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllNetworkAuthorizer,

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

@@ -101,6 +101,9 @@ public final class AuthConfig
 
         DatabaseDescriptor.setRoleManager(roleManager);
 
+        IDefaultRoleInitializer defaultRoleInitializer = authInstantiate(conf.default_role_initializer, IDefaultRoleInitializer.class, PasswordDefaultRoleInitializer.class);
+        DatabaseDescriptor.setDefaultRoleInitializer(defaultRoleInitializer);
+
         // authenticator
 
         IInternodeAuthenticator internodeAuthenticator = authInstantiate(conf.internode_authenticator,
@@ -140,6 +143,7 @@ public final class AuthConfig
         authenticator.validateConfiguration();
         authorizer.validateConfiguration();
         roleManager.validateConfiguration();
+        defaultRoleInitializer.validateConfiguration();
         networkAuthorizer.validateConfiguration();
         cidrAuthorizer.validateConfiguration();
         DatabaseDescriptor.getInternodeAuthenticator().validateConfiguration();

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

@@ -139,6 +139,12 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
      */
     static final ConsistencyLevel DEFAULT_SUPERUSER_CONSISTENCY_LEVEL = ConsistencyLevel.QUORUM;
 
+    /**
+     * Used when no default_role_initializer is configured, or when auth setup has not run, e.g. in tests which
+     * do not call {@link AuthConfig#applyAuth()}. Preserves the historical bootstrap behaviour.
+     */
+    private static final IDefaultRoleInitializer DEFAULT_ROLE_INITIALIZER = new PasswordDefaultRoleInitializer();
+
     // Transform a row in the AuthKeyspace.ROLES to a Role instance
     private static final Function<UntypedResultSet.Row, Role> ROW_TO_ROLE = row ->
     {
@@ -533,9 +539,7 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
         {
             if (!hasExistingRoles())
             {
-                QueryProcessor.process(createDefaultRoleQuery(),
+                defaultRoleInitializer().initialize();
-                                       consistencyForRoleWrite(DEFAULT_SUPERUSER_NAME));
-                logger.info("Created default superuser role '{}'", DEFAULT_SUPERUSER_NAME);
             }
         }
         catch (RequestExecutionException e)
@@ -558,11 +562,11 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
     @VisibleForTesting
     public static boolean hasExistingRoles() throws RequestExecutionException
     {
-        // Try looking up the 'cassandra' default role first, to avoid the range query if possible.
+        // Try looking up the configured default role first, to avoid the range query if possible.
-        String defaultSUQuery = String.format("SELECT * FROM %s.%s WHERE role = '%s'", SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES, DEFAULT_SUPERUSER_NAME);
+        String defaultRoleQuery = String.format("SELECT * FROM %s.%s WHERE role = '%s'", SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES, escape(defaultRoleInitializer().defaultRoleName()));
         String allUsersQuery = String.format("SELECT * FROM %s.%s LIMIT 1", SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES);
-        return !QueryProcessor.process(defaultSUQuery, ConsistencyLevel.ONE).isEmpty()
+        return !QueryProcessor.process(defaultRoleQuery, ConsistencyLevel.ONE).isEmpty()
-               || !QueryProcessor.process(defaultSUQuery, ConsistencyLevel.QUORUM).isEmpty()
+               || !QueryProcessor.process(defaultRoleQuery, ConsistencyLevel.QUORUM).isEmpty()
                || !QueryProcessor.process(allUsersQuery, ConsistencyLevel.QUORUM).isEmpty();
     }
 
@@ -771,16 +775,22 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
         throw new OverloadedException(failure);
     }
 
-    private static String hashpw(String password)
+    static String hashpw(String password)
     {
         return BCrypt.hashpw(password, PasswordSaltSupplier.get());
     }
 
-    private static String escape(String name)
+    static String escape(String name)
     {
         return StringUtils.replace(name, "'", "''");
     }
 
+    private static IDefaultRoleInitializer defaultRoleInitializer()
+    {
+        IDefaultRoleInitializer initializer = DatabaseDescriptor.getDefaultRoleInitializer();
+        return initializer == null ? DEFAULT_ROLE_INITIALIZER : initializer;
+    }
+
     private static ByteBuffer byteBuf(String str)
     {
         return UTF8Type.instance.decompose(str);
@@ -789,14 +799,14 @@ public class CassandraRoleManager implements IRoleManager, CassandraRoleManagerM
     /** Allows selective overriding of the consistency level for specific roles. */
     protected static ConsistencyLevel consistencyForRoleWrite(String role)
     {
-        return role.equals(DEFAULT_SUPERUSER_NAME) ?
+        return role.equals(defaultRoleInitializer().defaultRoleName()) ?
                DEFAULT_SUPERUSER_CONSISTENCY_LEVEL :
                CassandraAuthorizer.authWriteConsistencyLevel();
     }
 
     protected static ConsistencyLevel consistencyForRoleRead(String role)
     {
-        return role.equals(DEFAULT_SUPERUSER_NAME) ?
+        return role.equals(defaultRoleInitializer().defaultRoleName()) ?
                DEFAULT_SUPERUSER_CONSISTENCY_LEVEL :
                CassandraAuthorizer.authReadConsistencyLevel();
     }

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

@@ -0,0 +1,71 @@
+/*
+ * 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 org.apache.cassandra.exceptions.ConfigurationException;
+
+/**
+ * Creates the initial role on a cluster which has no roles yet, so that there is some
+ * identity available to authenticate as and grant permissions from. Selected via
+ * {@code default_role_initializer} option in cassandra.yaml and instantiated by
+ * {@link AuthConfig#applyAuth()}
+ *
+ * Implementations decide both what the role is called and how it is authenticated.
+ * See {@link PasswordDefaultRoleInitializer} which gives the role a password, and
+ * {@link MutualTlsDefaultRoleInitializer} which gives no password and instead
+ * maps a client certificate identity to itself.
+ */
+public interface IDefaultRoleInitializer
+{
+    /**
+     * Creates the default role. Called from {@link CassandraRoleManager#setup(boolean)} only after
+     * {@link CassandraRoleManager#hasExistingRoles()} has established that the cluster has no rules.
+     *
+     * Every node runs this independently during initial startup so implementations must write at
+     * {@link CassandraRoleManager#consistencyForRoleWrite(String)} to avoid concurrent duplicate creation
+     * and must use {@code USING TIMESTAMP 0} so that any operator changes to the role later supersede it.
+     *
+     * The caller retries on failure so this may be invoked more than once on a node: it must not fail
+     * @throws org.apache.cassandra.exceptions.RequestExecutionException if not enough nodes are available
+     * yet which the caller treats as a signal to reschedule
+     */
+    void initialize();
+
+    /**
+     * The name of the role {@link #initialize()} creates.
+     *
+     * The default role is a special case during startup: reads and writes of it are performed at
+     * {@link CassandraRoleManager#DEFAULT_SUPERUSER_CONSISTENCY_LEVEL} rather than at the configured auth
+     * consistency levels, and {@link CassandraRoleManager#hasExistingRoles()} looks it up by name before
+     * falling back to a range query. Both need to know the configured name, not assume
+     * {@link CassandraRoleManager#DEFAULT_SUPERUSER_NAME}.
+     */
+    String defaultRoleName();
+
+    /**
+     * Validates configuration of the IDefaultRoleInitializer implementation (if configurable).
+     *
+     * Called by {@link AuthConfig#applyAuth()} after the authenticator, authorizer and role manager have been
+     * set, so implementations may inspect those to reject combinations which would leave the cluster with no
+     * usable login.
+     *
+     * @throws ConfigurationException when there is a configuration error.
+     */
+    default void validateConfiguration() throws ConfigurationException {}
+}

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

@@ -0,0 +1,113 @@
+/*
+ * 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.Map;
+import java.util.Set;
+
+import com.google.common.base.Strings;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.schema.SchemaConstants;
+
+import static org.apache.cassandra.auth.CassandraRoleManager.consistencyForRoleWrite;
+import static org.apache.cassandra.auth.CassandraRoleManager.escape;
+
+public class MutualTlsDefaultRoleInitializer implements IDefaultRoleInitializer
+{
+    private static final Logger logger = LoggerFactory.getLogger(MutualTlsDefaultRoleInitializer.class);
+    static final String ROLE = "role";
+    static final String IDENTITY = "identity";
+
+    private static final Set<String> SUPPORTED_PARAMS = Set.of(ROLE, IDENTITY);
+    private final String role;
+    private final String identity;
+
+    public MutualTlsDefaultRoleInitializer(Map<String, String> parameters)
+    {
+        for (String params: parameters.keySet())
+        {
+            if (!SUPPORTED_PARAMS.contains(params))
+            {
+                throw new ConfigurationException(String.format("Unsupported parameter %s for %s, supported parameters are %s", params, getClass().getSimpleName(), SUPPORTED_PARAMS));
+            }
+        }
+        role = parameters.get(ROLE);
+        identity = parameters.get(IDENTITY);
+    }
+
+    @Override
+    public void initialize()
+    {
+        QueryProcessor.process(String.format("INSERT INTO %s.%s (role, is_superuser, can_login) " +
+                                             "VALUES ('%s', true, true) USING TIMESTAMP 0",
+                                             SchemaConstants.AUTH_KEYSPACE_NAME,
+                                             AuthKeyspace.ROLES,
+                                             escape(role)),
+                               consistencyForRoleWrite(role));
+
+        QueryProcessor.process(String.format("INSERT INTO %s.%s (identity, role) " +
+                                             "VALUES ('%s', '%s') USING TIMESTAMP 0",
+                                             SchemaConstants.AUTH_KEYSPACE_NAME,
+                                             AuthKeyspace.IDENTITY_TO_ROLES,
+                                             escape(identity),
+                                             escape(role)),
+                               consistencyForRoleWrite(role));
+
+        logger.info("Created passwordless default superuser role '{}' for identity '{}'", role, identity);
+    }
+
+    @Override
+    public String defaultRoleName()
+    {
+        return role;
+    }
+
+    @Override
+    public void validateConfiguration() throws ConfigurationException
+    {
+        if (Strings.isNullOrEmpty(role))
+            throw new ConfigurationException(String.format("%s requires a non-empty '%s' parameter",
+                                                           getClass().getSimpleName(), ROLE));
+
+        if (Strings.isNullOrEmpty(identity))
+            throw new ConfigurationException(String.format("%s requires a non-empty '%s' parameter",
+                                                           getClass().getSimpleName(), IDENTITY));
+
+        // The role this creates has no password, so an authenticator which cannot authenticate by certificate
+        // would leave a freshly bootstrapped cluster with no way to log in at all.
+        IAuthenticator authenticator = DatabaseDescriptor.getAuthenticator();
+        Set<IAuthenticator.AuthenticationMode> modes = authenticator.getSupportedAuthenticationModes();
+        if (authenticator.requireAuthentication() && !modes.isEmpty() && !modes.contains(IAuthenticator.AuthenticationMode.MTLS))
+        {
+            throw new ConfigurationException(String.format("%s creates a role with no password, which %s cannot " +
+                                                           "authenticate (supported modes: %s). Configure an " +
+                                                           "authenticator supporting mutual TLS, such as %s.",
+                                                           getClass().getSimpleName(),
+                                                           authenticator.getClass().getSimpleName(),
+                                                           modes,
+                                                           MutualTlsAuthenticator.class.getSimpleName()));
+        }
+    }
+}

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

@@ -0,0 +1,107 @@
+/*
+ * 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.Map;
+import java.util.Set;
+
+import com.google.common.base.Strings;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.schema.SchemaConstants;
+
+import static org.apache.cassandra.auth.CassandraRoleManager.DEFAULT_SUPERUSER_NAME;
+import static org.apache.cassandra.auth.CassandraRoleManager.DEFAULT_SUPERUSER_PASSWORD;
+import static org.apache.cassandra.auth.CassandraRoleManager.consistencyForRoleWrite;
+import static org.apache.cassandra.auth.CassandraRoleManager.escape;
+import static org.apache.cassandra.auth.CassandraRoleManager.hashpw;
+
+/**
+ * Creates the default role with a password, so that it can be authenticated with
+ * {@link PasswordAuthenticator}. This is the default {@link IDefaultRoleInitializer} and reproduces the
+ * historical bootstrap behaviour of creating a {@code cassandra} superuser whose password is also
+ * {@code cassandra}.
+ *
+ * Because that password is a well known constant, deployments which can authenticate by other means should
+ * prefer an initializer which does not create a password at all, such as
+ * {@link MutualTlsDefaultRoleInitializer}. Deployments which do use this initializer should rotate or drop the
+ * created role before the native transport is reachable.
+ */
+public class PasswordDefaultRoleInitializer implements IDefaultRoleInitializer
+{
+    private static final Logger logger = LoggerFactory.getLogger(PasswordDefaultRoleInitializer.class);
+    static final String ROLE = "role";
+    static final String PASSWORD = "password";
+
+    private static final Set<String> SUPPORTED_PARAMS = Set.of(ROLE, PASSWORD);
+
+    private final String role;
+    private final String password;
+
+    public PasswordDefaultRoleInitializer()
+    {
+        this(Map.of());
+    }
+
+    public PasswordDefaultRoleInitializer(Map<String, String> parameters)
+    {
+        for (String param: parameters.keySet())
+        {
+            if (!SUPPORTED_PARAMS.contains(param))
+                throw new ConfigurationException(String.format("Unsupported parameter '%s' for %s, supported parameters are %s", param, getClass().getSimpleName(), SUPPORTED_PARAMS));
+        }
+
+        role = parameters.getOrDefault(ROLE, DEFAULT_SUPERUSER_NAME);
+        password = parameters.getOrDefault(PASSWORD, DEFAULT_SUPERUSER_PASSWORD);
+    }
+
+    @Override
+    public void initialize()
+    {
+        QueryProcessor.process(String.format("INSERT INTO %s.%s (role, is_superuser, can_login, salted_hash) "
+                                             +
+                                             "VALUES ('%s', true, true, '%s') USING TIMESTAMP 0",
+                                             SchemaConstants.AUTH_KEYSPACE_NAME,
+                                             AuthKeyspace.ROLES,
+                                             escape(role),
+                                             escape(hashpw(password))),
+                               consistencyForRoleWrite(role));
+        logger.info("Created default superuser role '{}'", role);
+    }
+
+    @Override
+    public String defaultRoleName()
+    {
+        return role;
+    }
+
+    @Override
+    public void validateConfiguration() throws ConfigurationException
+    {
+        if (Strings.isNullOrEmpty(role))
+            throw new ConfigurationException(String.format("%s requires a non-empty %s parameter", getClass().getSimpleName(), ROLE));
+
+        if (Strings.isNullOrEmpty(password))
+            throw new ConfigurationException(String.format("%s requires a non-empty %s parameter", getClass().getSimpleName(), PASSWORD));
+    }
+}

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

@@ -86,6 +86,7 @@ public class Config
     public ParameterizedClass authenticator;
     public ParameterizedClass authorizer;
     public ParameterizedClass role_manager;
+    public ParameterizedClass default_role_initializer;
     public ParameterizedClass crypto_provider;
     public ParameterizedClass network_authorizer;
     public ParameterizedClass cidr_authorizer;

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

@@ -77,6 +77,7 @@ import org.apache.cassandra.auth.AuthConfig;
 import org.apache.cassandra.auth.IAuthenticator;
 import org.apache.cassandra.auth.IAuthorizer;
 import org.apache.cassandra.auth.ICIDRAuthorizer;
+import org.apache.cassandra.auth.IDefaultRoleInitializer;
 import org.apache.cassandra.auth.IInternodeAuthenticator;
 import org.apache.cassandra.auth.INetworkAuthorizer;
 import org.apache.cassandra.auth.IRoleManager;
@@ -240,6 +241,7 @@ public class DatabaseDescriptor
     // Don't initialize the role manager until applying config. The options supported by CassandraRoleManager
     // depend on the configured IAuthenticator, so defer creating it until that's been set.
     private static IRoleManager roleManager;
+    private static IDefaultRoleInitializer defaultRoleInitializer;
 
     private static long preparedStatementsCacheSizeInMiB;
 
@@ -2226,6 +2228,16 @@ public class DatabaseDescriptor
         DatabaseDescriptor.roleManager = roleManager;
     }
 
+    public static IDefaultRoleInitializer getDefaultRoleInitializer()
+    {
+        return defaultRoleInitializer;
+    }
+
+    public static void setDefaultRoleInitializer(IDefaultRoleInitializer defaultRoleInitializer)
+    {
+        DatabaseDescriptor.defaultRoleInitializer = defaultRoleInitializer;
+    }
+
     public static int getPermissionsValidity()
     {
         return conf.permissions_validity.toMilliseconds();

test/distributed/org/apache/cassandra/distributed/test/auth/MutualTlsDefaultRoleInitializerTest.java

@@ -0,0 +1,176 @@
+/*
+ * 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.distributed.test.auth;
+
+import java.net.InetAddress;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Map;
+
+import com.datastax.driver.core.Session;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import org.apache.cassandra.auth.AuthKeyspace;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableCallable;
+import org.apache.cassandra.distributed.test.JavaDriverUtils;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.utils.tls.CertificateBuilder;
+import org.apache.cassandra.utils.tls.CertificateBundle;
+
+import static org.apache.cassandra.transport.TlsTestUtils.CLIENT_SPIFFE_IDENTITY;
+import static org.apache.cassandra.transport.TlsTestUtils.SERVER_KEYSTORE_PASSWORD;
+import static org.apache.cassandra.transport.TlsTestUtils.SERVER_TRUSTSTORE_PASSWORD;
+import static org.apache.cassandra.transport.TlsTestUtils.generateClientCertificate;
+import static org.apache.cassandra.transport.TlsTestUtils.getSSLOptions;
+import static org.apache.cassandra.transport.TlsTestUtils.withAuthenticatedSession;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Verifies that {@code default_role_initializer: MutualTlsDefaultRoleInitializer} bootstraps a passwordless
+ * superuser role and its identity mapping without any post-startup CQL step, that a client presenting a
+ * matching certificate can authenticate as that role with superuser access, and that password authentication
+ * against the same role fails cleanly since it has no {@code salted_hash}.
+ */
+public class MutualTlsDefaultRoleInitializerTest extends TestBaseImpl
+{
+    private static final String TEST_ROLE = "cassandra_mtls_bootstrap_test_role";
+
+    @ClassRule
+    public static TemporaryFolder tempFolder = new TemporaryFolder();
+
+    private static ICluster<IInvokableInstance> CLUSTER;
+    private static CertificateBundle CA;
+    private static Path truststorePath;
+    private static Path clientKeystorePath;
+
+    @BeforeClass
+    public static void setupClass() throws Exception
+    {
+        Cluster.Builder builder = Cluster.build(1).withDynamicPortAllocation(true);
+
+        CA = new CertificateBuilder().subject("CN=Apache Cassandra Root CA, OU=Certification Authority, O=Unknown, C=Unknown")
+                                     .alias("fakerootca")
+                                     .isCertificateAuthority(true)
+                                     .buildSelfSigned();
+
+        truststorePath = CA.toTempKeyStorePath(tempFolder.getRoot().toPath(),
+                                               SERVER_TRUSTSTORE_PASSWORD.toCharArray(),
+                                               SERVER_TRUSTSTORE_PASSWORD.toCharArray());
+
+        CertificateBundle serverKeystore = new CertificateBuilder().subject("CN=Apache Cassandra, OU=ssl_test, O=Unknown, L=Unknown, ST=Unknown, C=Unknown")
+                                                                   .addSanDnsName(InetAddress.getLocalHost().getCanonicalHostName())
+                                                                   .addSanDnsName(InetAddress.getLocalHost().getHostName())
+                                                                   .buildIssuedBy(CA);
+
+        Path serverKeystorePath = serverKeystore.toTempKeyStorePath(tempFolder.getRoot().toPath(),
+                                                                     SERVER_KEYSTORE_PASSWORD.toCharArray(),
+                                                                     SERVER_KEYSTORE_PASSWORD.toCharArray());
+
+        builder.withConfig(c -> c.set("authenticator.class_name", "org.apache.cassandra.auth.MutualTlsWithPasswordFallbackAuthenticator")
+                                 .set("authenticator.parameters", Collections.singletonMap("validator_class_name", "org.apache.cassandra.auth.SpiffeCertificateValidator"))
+                                 .set("role_manager", "CassandraRoleManager")
+                                 .set("authorizer", "CassandraAuthorizer")
+                                 .set("default_role_initializer.class_name", "org.apache.cassandra.auth.MutualTlsDefaultRoleInitializer")
+                                 .set("default_role_initializer.parameters", Map.of("role", TEST_ROLE, "identity", CLIENT_SPIFFE_IDENTITY))
+                                 .set("client_encryption_options.enabled", "true")
+                                 .set("client_encryption_options.require_client_auth", "optional")
+                                 .set("client_encryption_options.keystore", serverKeystorePath.toString())
+                                 .set("client_encryption_options.keystore_password", SERVER_KEYSTORE_PASSWORD)
+                                 .set("client_encryption_options.truststore", truststorePath.toString())
+                                 .set("client_encryption_options.truststore_password", SERVER_TRUSTSTORE_PASSWORD)
+                                 .set("client_encryption_options.require_endpoint_verification", "false")
+                                 .with(Feature.NATIVE_PROTOCOL, Feature.GOSSIP, Feature.NETWORK));
+        CLUSTER = builder.start();
+
+        clientKeystorePath = generateClientCertificate(null, tempFolder.getRoot(), CA);
+    }
+
+    @AfterClass
+    public static void teardown() throws Exception
+    {
+        if (CLUSTER != null)
+            CLUSTER.close();
+    }
+
+    @Test
+    public void testDefaultRoleAndIdentityCreatedAtBootstrap()
+    {
+        String identity = CLIENT_SPIFFE_IDENTITY;
+
+        Object[] roleRow = CLUSTER.get(1).callOnInstance((SerializableCallable<Object[]>) () -> {
+            UntypedResultSet result = QueryProcessor.executeInternal(
+                String.format("SELECT is_superuser, can_login, salted_hash FROM %s.%s WHERE role = '%s'",
+                              SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES, TEST_ROLE));
+            if (result.isEmpty())
+                return null;
+            UntypedResultSet.Row row = result.one();
+            return new Object[]{ row.getBoolean("is_superuser"), row.getBoolean("can_login"), row.has("salted_hash") };
+        });
+
+        assertThat(roleRow).isNotNull();
+        assertThat((Boolean) roleRow[0]).as("is_superuser").isTrue();
+        assertThat((Boolean) roleRow[1]).as("can_login").isTrue();
+        assertThat((Boolean) roleRow[2]).as("has salted_hash").isFalse();
+
+        String mappedRole = CLUSTER.get(1).callOnInstance((SerializableCallable<String>) () -> {
+            UntypedResultSet result = QueryProcessor.executeInternal(
+                String.format("SELECT role FROM %s.%s WHERE identity = '%s'",
+                              SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.IDENTITY_TO_ROLES, identity));
+            Assert.assertNotNull(result);
+            return result.isEmpty() ? null : result.one().getString("role");
+        });
+
+        assertThat(mappedRole).isEqualTo(TEST_ROLE);
+    }
+
+    @Test
+    public void testCertificateAuthenticationGrantsSuperuser() throws Exception
+    {
+        try (com.datastax.driver.core.Cluster c = JavaDriverUtils.create(CLUSTER, null, b -> b.withSSL(getSSLOptions(clientKeystorePath, truststorePath)));
+             Session session = c.connect())
+        {
+            // system_auth.roles is a protected resource (CassandraRoleManager#protectedResources); reading it
+            // without ever being granted SELECT proves this connection authenticated with superuser status.
+            com.datastax.driver.core.ResultSet rows = session.execute("SELECT role FROM system_auth.roles WHERE role = ?", TEST_ROLE);
+            assertThat(rows.one().getString("role")).isEqualTo(TEST_ROLE);
+        }
+    }
+
+    @Test
+    public void testPasswordAuthenticationFailsCleanly() throws Exception
+    {
+        assertThatThrownBy(() -> withAuthenticatedSession(CLUSTER.get(1), TEST_ROLE, "irrelevant-password", session -> {
+        }, getSSLOptions(null, truststorePath)))
+            .isInstanceOf(com.datastax.driver.core.exceptions.AuthenticationException.class);
+    }
+}

test/distributed/org/apache/cassandra/distributed/test/auth/PasswordDefaultRoleInitializerTest.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.distributed.test.auth;
+
+import java.net.InetAddress;
+import java.util.function.Consumer;
+
+import com.datastax.driver.core.Cluster.Builder;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.policies.LoadBalancingPolicy;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.auth.AuthKeyspace;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.ICluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableCallable;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.Auth;
+import org.apache.cassandra.distributed.util.SingleHostLoadBalancingPolicy;
+import org.apache.cassandra.schema.SchemaConstants;
+
+import static org.apache.cassandra.auth.CassandraRoleManager.DEFAULT_SUPERUSER_NAME;
+import static org.apache.cassandra.auth.CassandraRoleManager.DEFAULT_SUPERUSER_PASSWORD;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Regression test for the {@code default_role_initializer} refactor (see {@link org.apache.cassandra.auth.IDefaultRoleInitializer}):
+ * when the option is left unconfigured, bootstrap must still fall back to {@link org.apache.cassandra.auth.PasswordDefaultRoleInitializer}
+ * and create the classic {@code cassandra}/{@code cassandra} superuser exactly as it always has, so deployments and
+ * tests that rely on the historical default (e.g. {@code CQLTester}) keep working unchanged.
+ */
+public class PasswordDefaultRoleInitializerTest extends TestBaseImpl
+{
+    private static ICluster<IInvokableInstance> CLUSTER;
+
+    @BeforeClass
+    public static void setupClass() throws Exception
+    {
+        CLUSTER = Cluster.build(1)
+                        .withConfig(conf -> conf.with(Feature.GOSSIP, Feature.NATIVE_PROTOCOL)
+                                                .set("authenticator", "PasswordAuthenticator")
+                                                .set("authorizer", "CassandraAuthorizer")
+                                                .set("role_manager", "CassandraRoleManager"))
+                        .start();
+    }
+
+    @AfterClass
+    public static void teardown() throws Exception
+    {
+        if (CLUSTER != null)
+            CLUSTER.close();
+    }
+
+    @Test
+    public void testClassicSuperuserBootstrappedByDefault()
+    {
+        Object[] roleRow = CLUSTER.get(1).callOnInstance((SerializableCallable<Object[]>) () -> {
+            UntypedResultSet result = QueryProcessor.executeInternal(
+                String.format("SELECT is_superuser, can_login, salted_hash FROM %s.%s WHERE role = '%s'",
+                              SchemaConstants.AUTH_KEYSPACE_NAME, AuthKeyspace.ROLES, DEFAULT_SUPERUSER_NAME));
+            if (result.isEmpty())
+                return null;
+            UntypedResultSet.Row row = result.one();
+            return new Object[]{ row.getBoolean("is_superuser"), row.getBoolean("can_login"), row.has("salted_hash") };
+        });
+
+        assertThat(roleRow).isNotNull();
+        assertThat((Boolean) roleRow[0]).as("is_superuser").isTrue();
+        assertThat((Boolean) roleRow[1]).as("can_login").isTrue();
+        assertThat((Boolean) roleRow[2]).as("has salted_hash").isTrue();
+
+        // and the credential actually works end-to-end, not just the raw row contents
+        withAuthenticatedSession(CLUSTER.get(1), DEFAULT_SUPERUSER_NAME, DEFAULT_SUPERUSER_PASSWORD, session -> {
+            com.datastax.driver.core.ResultSet rows = session.execute("SELECT role FROM system_auth.roles WHERE role = ?", DEFAULT_SUPERUSER_NAME);
+            assertThat(rows.one().getString("role")).isEqualTo(DEFAULT_SUPERUSER_NAME);
+        });
+    }
+
+    // No client_encryption_options are configured for this cluster, so unlike TlsTestUtils#withAuthenticatedSession
+    // (which always requires SSLOptions) this connects in plaintext, matching ColumnMaskTest's local helper.
+    private static void withAuthenticatedSession(IInvokableInstance instance, String username, String password, Consumer<Session> consumer)
+    {
+        Auth.waitForExistingRoles(instance);
+
+        InetAddress address = instance.broadcastAddress().getAddress();
+        LoadBalancingPolicy lbc = new SingleHostLoadBalancingPolicy(address);
+
+        Builder builder = com.datastax.driver.core.Cluster.builder()
+                                                          .addContactPoints(address)
+                                                          .withLoadBalancingPolicy(lbc)
+                                                          .withCredentials(username, password);
+
+        try (com.datastax.driver.core.Cluster cluster = builder.build(); Session session = cluster.connect())
+        {
+            consumer.accept(session);
+        }
+    }
+}

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

@@ -23,6 +23,7 @@ import java.security.cert.Certificate;
 import java.security.cert.CertificateException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Map;
 
 import org.junit.After;
 import org.junit.Before;
@@ -164,6 +165,29 @@ public class AuthConfigTest
         assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(authenticator.getAlterableRoleOptions()));
     }
 
+    @Test
+    public void testNewInstanceForMutualTlsDefaultRoleInitializer()
+    {
+        Config config = load("cassandra-mtls.yaml");
+        config.default_role_initializer = new ParameterizedClass("org.apache.cassandra.auth.MutualTlsDefaultRoleInitializer",
+                                                                  Map.of("role", "cassandra", "identity", "spiffe1"));
+        DatabaseDescriptor.unsafeDaemonInitialization(()->config);
+
+        assertThat(DatabaseDescriptor.getDefaultRoleInitializer()).isInstanceOf(MutualTlsDefaultRoleInitializer.class);
+        assertThat(DatabaseDescriptor.getDefaultRoleInitializer().defaultRoleName()).isEqualTo("cassandra");
+    }
+
+    @Test
+    public void testMutualTlsDefaultRoleInitializerRejectedWithIncompatibleAuthenticator()
+    {
+        Config config = load("cassandra-passwordauth.yaml");
+        config.default_role_initializer = new ParameterizedClass("org.apache.cassandra.auth.MutualTlsDefaultRoleInitializer",
+                                                                  Map.of("role", "cassandra", "identity", "spiffe1"));
+        assertThatThrownBy(() -> DatabaseDescriptor.unsafeDaemonInitialization(()->config))
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("creates a role with no password");
+    }
+
     private static final String PROBE = ClassLoadingTestNonAssignable.class.getName();
 
     private static Config baseConfig()
@@ -209,6 +233,14 @@ public class AuthConfigTest
         assertApplyAuthRejectsProbe();
     }
 
+    @Test
+    public void testDefaultRoleInitializerWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.default_role_initializer = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
     @Test
     public void testInternodeAuthenticatorWrongTypeRejectedWithoutInitializing()
     {

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

@@ -0,0 +1,116 @@
+/*
+ * 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.Map;
+
+import org.junit.Test;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Unit tests for {@link PasswordDefaultRoleInitializer} and {@link MutualTlsDefaultRoleInitializer} that don't
+ * require a running node: unsupported-parameter rejection at construction time, and the parameter-presence checks
+ * in {@link IDefaultRoleInitializer#validateConfiguration()}. The authenticator-compatibility half of
+ * {@link MutualTlsDefaultRoleInitializer#validateConfiguration()} needs a configured {@code DatabaseDescriptor}
+ * authenticator and is covered by {@link AuthConfigTest} instead.
+ */
+public class DefaultRoleInitializerTest
+{
+    @Test
+    public void passwordInitializerRejectsUnsupportedParameter()
+    {
+        assertThatThrownBy(() -> new PasswordDefaultRoleInitializer(Map.of("bogus", "x")))
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("Unsupported parameter");
+    }
+
+    @Test
+    public void passwordInitializerDefaultsPassValidation()
+    {
+        new PasswordDefaultRoleInitializer().validateConfiguration();
+        new PasswordDefaultRoleInitializer(Collections.emptyMap()).validateConfiguration();
+    }
+
+    @Test
+    public void passwordInitializerRejectsEmptyRole()
+    {
+        assertThatThrownBy(() -> new PasswordDefaultRoleInitializer(Map.of("role", "", "password", "x")).validateConfiguration())
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("role");
+    }
+
+    @Test
+    public void passwordInitializerRejectsEmptyPassword()
+    {
+        assertThatThrownBy(() -> new PasswordDefaultRoleInitializer(Map.of("role", "cassandra", "password", "")).validateConfiguration())
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("password");
+    }
+
+    @Test
+    public void passwordInitializerDefaultRoleNameMatchesConfiguredRole()
+    {
+        PasswordDefaultRoleInitializer initializer = new PasswordDefaultRoleInitializer(Map.of("role", "myrole", "password", "x"));
+        assertThat(initializer.defaultRoleName()).isEqualTo("myrole");
+    }
+
+    @Test
+    public void mutualTlsInitializerRejectsUnsupportedParameter()
+    {
+        assertThatThrownBy(() -> new MutualTlsDefaultRoleInitializer(Map.of("bogus", "x")))
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("Unsupported parameter");
+    }
+
+    @Test
+    public void mutualTlsInitializerRejectsMissingRole()
+    {
+        assertThatThrownBy(() -> new MutualTlsDefaultRoleInitializer(Map.of("identity", "spiffe1")).validateConfiguration())
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("role");
+    }
+
+    @Test
+    public void mutualTlsInitializerRejectsMissingIdentity()
+    {
+        assertThatThrownBy(() -> new MutualTlsDefaultRoleInitializer(Map.of("role", "cassandra")).validateConfiguration())
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("identity");
+    }
+
+    @Test
+    public void mutualTlsInitializerRejectsEmptyRoleAndIdentity()
+    {
+        assertThatThrownBy(() -> new MutualTlsDefaultRoleInitializer(Map.of("role", "", "identity", "")).validateConfiguration())
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("role");
+    }
+
+    @Test
+    public void mutualTlsInitializerDefaultRoleNameMatchesConfiguredRole()
+    {
+        MutualTlsDefaultRoleInitializer initializer = new MutualTlsDefaultRoleInitializer(Map.of("role", "cassandra", "identity", "spiffe1"));
+        assertThat(initializer.defaultRoleName()).isEqualTo("cassandra");
+    }
+}

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

@@ -20,6 +20,7 @@ package org.apache.cassandra.auth;
 
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -132,6 +133,33 @@ public class RolesTest
         Assert.assertEquals(nonPrivWriteLevel, DatabaseDescriptor.getAuthWriteConsistencyLevel());
     }
 
+    @Test
+    public void confirmSuperUserConsistencyWithConfiguredDefaultRoleName()
+    {
+        IDefaultRoleInitializer previous = DatabaseDescriptor.getDefaultRoleInitializer();
+        String customRole = "cassandra_mtls_custom_test_role";
+        try
+        {
+            DatabaseDescriptor.setDefaultRoleInitializer(new MutualTlsDefaultRoleInitializer(Map.of("role", customRole, "identity", "spiffe1")));
+
+            ConsistencyLevel readLevel = CassandraRoleManager.consistencyForRoleRead(customRole);
+            Assert.assertEquals(CassandraRoleManager.DEFAULT_SUPERUSER_CONSISTENCY_LEVEL, readLevel);
+
+            ConsistencyLevel writeLevel = CassandraRoleManager.consistencyForRoleWrite(customRole);
+            Assert.assertEquals(CassandraRoleManager.DEFAULT_SUPERUSER_CONSISTENCY_LEVEL, writeLevel);
+
+            ConsistencyLevel legacyReadLevel = CassandraRoleManager.consistencyForRoleRead(CassandraRoleManager.DEFAULT_SUPERUSER_NAME);
+            Assert.assertEquals(legacyReadLevel, DatabaseDescriptor.getAuthReadConsistencyLevel());
+
+            ConsistencyLevel legacyWriteLevel = CassandraRoleManager.consistencyForRoleWrite(CassandraRoleManager.DEFAULT_SUPERUSER_NAME);
+            Assert.assertEquals(legacyWriteLevel, DatabaseDescriptor.getAuthWriteConsistencyLevel());
+        }
+        finally
+        {
+            DatabaseDescriptor.setDefaultRoleInitializer(previous);
+        }
+    }
+
     @Test
     public void testSuperUsers()
     {

test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java

@@ -74,6 +74,7 @@ public class DatabaseDescriptorRefTest
     "org.apache.cassandra.auth.IAuthorizer",
     "org.apache.cassandra.auth.ICIDRAuthorizer",
     "org.apache.cassandra.auth.ICIDRAuthorizer$CIDRAuthorizerMode",
+    "org.apache.cassandra.auth.IDefaultRoleInitializer",
     "org.apache.cassandra.auth.IInternodeAuthenticator",
     "org.apache.cassandra.auth.INetworkAuthorizer",
     "org.apache.cassandra.auth.IRoleManager",