CASSANDRA-19385: ALTER ROLE WITH LOGIN=FALSE and REVOKE ROLE do not disconnect existing users

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

@@ -28,8 +28,10 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.function.IntSupplier;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -42,6 +44,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.CQLStatement;
@@ -59,6 +62,7 @@
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.StorageProxy;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.transport.Dispatcher;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -181,6 +185,7 @@ public void setup(boolean asyncRoleSetup)
             setupDefaultRole();
             return null;
         });
+        scheduleDisconnectInvalidRoleTask();
     }
 
     @Override
@@ -301,6 +306,7 @@ public void dropRole(AuthenticatedUser performer, RoleResource role) throws Requ
                 consistencyForRoleWrite(role.getRoleName()));
         removeAllMembers(role.getRoleName());
         removeAllIdentitiesOfRole(role.getRoleName());
+        disconnect(role);
     }
 
     public void alterRole(AuthenticatedUser performer, RoleResource role, RoleOptions options)
@@ -316,6 +322,9 @@ public void alterRole(AuthenticatedUser performer, RoleResource role, RoleOption
                                   assignments,
                                   escape(role.getRoleName())),
                     consistencyForRoleWrite(role.getRoleName()));
+
+            if (!options.getLogin().orElse(false))
+                disconnect(role);
         }
     }
 
@@ -732,4 +741,36 @@ public Supplier<Map<RoleResource, Set<Role>>> bulkLoader()
             return entries;
         };
     }
+
+    protected void scheduleDisconnectInvalidRoleTask()
+    {
+        ScheduledExecutorPlus executor = ScheduledExecutors.optionalTasks;
+
+        // REVIEW: I'm looking for input on how users would want to configure this. I can see applicability for:
+        // - limiting connections to live for a maximum duration + jitter
+        // - scheduling invalid role disconnects periodically + jitter
+        // - disabling disconnects (previous behavior)
+
+        // Delay between executions, with jitter
+        IntSupplier delayMillis = () -> (int) TimeUnit.HOURS.toMillis(4) + ThreadLocalRandom.current().nextInt(0, (int) TimeUnit.HOURS.toMillis(1));
+
+        executor.schedule(() -> {
+            try
+            {
+                StorageService.instance.disconnectInvalidRoles();
+            }
+            catch (Exception e)
+            {
+                logger.warn("Failed to disconnect invalid roles", e);
+            }
+
+            executor.schedule(this::scheduleDisconnectInvalidRoleTask, delayMillis.getAsInt(), TimeUnit.MILLISECONDS);
+        }, delayMillis.getAsInt(), TimeUnit.MILLISECONDS);
+    }
+
+    public void disconnect(RoleResource role)
+    {
+        logger.info("Disconnecting role {}", role);
+        StorageService.instance.disconnect(user -> user.getName().equals(role.getRoleName()));
+    }
 }

src/java/org/apache/cassandra/service/CassandraDaemon.java

@@ -31,6 +31,7 @@
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 import javax.management.StandardMBean;
 import javax.management.remote.JMXConnectorServer;
@@ -45,6 +46,7 @@
 import com.codahale.metrics.SharedMetricRegistries;
 import org.apache.cassandra.audit.AuditLogManager;
 import org.apache.cassandra.auth.AuthCacheService;
+import org.apache.cassandra.auth.AuthenticatedUser;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
 import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -893,6 +895,11 @@ public void clearConnectionHistory()
         nativeTransportService.clearConnectionHistory();
     }
 
+    public void disconnectUser(Predicate<AuthenticatedUser> userPredicate)
+    {
+        nativeTransportService.disconnect(userPredicate);
+    }
+
     private void exitOrFail(int code, String message)
     {
         exitOrFail(code, message, null);

src/java/org/apache/cassandra/service/NativeTransportService.java

@@ -19,6 +19,7 @@
 
 import java.net.InetAddress;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 
 import com.google.common.annotations.VisibleForTesting;
 
@@ -30,6 +31,7 @@
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.util.Version;
+import org.apache.cassandra.auth.AuthenticatedUser;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.EncryptionOptions;
 import org.apache.cassandra.metrics.ClientMetrics;
@@ -164,4 +166,9 @@ public void clearConnectionHistory()
     {
         server.clearConnectionHistory();
     }
+
+    public void disconnect(Predicate<AuthenticatedUser> userPredicate)
+    {
+        server.disconnect(userPredicate);
+    }
 }

src/java/org/apache/cassandra/service/StorageService.java

@@ -83,6 +83,7 @@
 import org.apache.cassandra.audit.AuditLogOptions;
 import org.apache.cassandra.auth.AuthCacheService;
 import org.apache.cassandra.auth.AuthSchemaChangeListener;
+import org.apache.cassandra.auth.AuthenticatedUser;
 import org.apache.cassandra.batchlog.BatchlogManager;
 import org.apache.cassandra.concurrent.ExecutorLocals;
 import org.apache.cassandra.concurrent.FutureTask;
@@ -4969,6 +4970,19 @@ public void clearConnectionHistory()
         daemon.clearConnectionHistory();
         logger.info("Cleared connection history");
     }
+
+    public void disconnect(Predicate<AuthenticatedUser> userPredicate)
+    {
+        logger.info("Disconnecting users matching predicate {}", userPredicate);
+        daemon.disconnectUser(userPredicate);
+    }
+
+    public void disconnectInvalidRoles()
+    {
+        logger.info("Disconnecting invalid roles");
+        daemon.disconnectUser(user -> !user.canLogin());
+    }
+
     public void disableAuditLog()
     {
         AuditLogManager.instance.disableAuditLog();

src/java/org/apache/cassandra/transport/Server.java

@@ -211,6 +211,11 @@ private void close(boolean force)
         logger.info("Stop listening for CQL clients");
     }
 
+    public void disconnect(Predicate<AuthenticatedUser> userPredicate)
+    {
+        connectionTracker.disconnectByUser(userPredicate);
+    }
+
     public static class Builder
     {
         private EventLoopGroup workerGroup;
@@ -383,6 +388,23 @@ Map<String, Integer> countConnectedClientsByUser()
             return result;
         }
 
+        void disconnectByUser(Predicate<AuthenticatedUser> userPredicate)
+        {
+            for (Channel c : allChannels)
+            {
+                Connection connection = c.attr(Connection.attributeKey).get();
+                if (connection instanceof ServerConnection)
+                {
+                    ServerConnection conn = (ServerConnection) connection;
+                    AuthenticatedUser user = conn.getClientState().getUser();
+                    if (userPredicate.test(user))
+                    {
+                        logger.info("Closing channel with user {}", user);
+                        connection.channel().close();
+                    }
+                }
+            }
+        }
     }
 
     private static class LatestEvent

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

@@ -0,0 +1,142 @@
+/*
+ * 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.util.function.Consumer;
+
+import org.junit.Test;
+
+import com.datastax.driver.core.PlainTextAuthProvider;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.policies.ConstantReconnectionPolicy;
+import com.datastax.driver.core.policies.ReconnectionPolicy;
+import org.apache.cassandra.auth.AuthenticatedUser;
+import org.apache.cassandra.auth.IRoleManager;
+import org.apache.cassandra.auth.RoleOptions;
+import org.apache.cassandra.auth.RoleResource;
+import org.apache.cassandra.auth.Roles;
+import org.apache.cassandra.config.DatabaseDescriptor;
+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.test.JavaDriverUtils;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.service.StorageService;
+import org.assertj.core.api.Assertions;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+
+public class RoleRevocationTest extends TestBaseImpl
+{
+    // Ensure both the coordinator of the DDL and a replica both change connection state accordingly
+    private static final int CLUSTER_SIZE = 2;
+    private static final String USERNAME = "revoke_me";
+    private static final String PASSWORD = "i_deserve_disconnection";
+    private static final PlainTextAuthProvider CLIENT_AUTH_PROVIDER = new PlainTextAuthProvider(USERNAME, PASSWORD);
+    private static final ReconnectionPolicy RECONNECTION_POLICY = new ConstantReconnectionPolicy(100);
+
+    public static ICluster<IInvokableInstance> cluster() throws Exception
+    {
+        Cluster.Builder builder = Cluster.build(CLUSTER_SIZE);
+
+        builder.withConfig(c -> c.set("authenticator.class_name", "org.apache.cassandra.auth.PasswordAuthenticator")
+                                 .set("role_manager", "CassandraRoleManager")
+                                 .set("authorizer", "CassandraAuthorizer")
+                                 .with(Feature.NETWORK, Feature.NATIVE_PROTOCOL, Feature.GOSSIP));
+        ICluster<IInvokableInstance> cluster = builder.start();
+
+        cluster.get(1).runOnInstance(() -> {
+            RoleOptions opts = new RoleOptions();
+            opts.setOption(IRoleManager.Option.PASSWORD, PASSWORD);
+            opts.setOption(IRoleManager.Option.LOGIN, true);
+            DatabaseDescriptor.getRoleManager().createRole(AuthenticatedUser.SYSTEM_USER, RoleResource.role(USERNAME), opts);
+        });
+
+        return cluster;
+    }
+
+    @Test
+    public void alterRoleLoginFalse() throws Exception
+    {
+        test(instance -> {
+            instance.runOnInstance(() -> {
+                RoleOptions opts = new RoleOptions();
+                opts.setOption(IRoleManager.Option.LOGIN, false);
+                DatabaseDescriptor.getRoleManager().alterRole(AuthenticatedUser.SYSTEM_USER, RoleResource.role(USERNAME), opts);
+            });
+        });
+    }
+
+    @Test
+    public void dropRole() throws Exception
+    {
+        test(instance -> {
+            instance.runOnInstance(() -> {
+                DatabaseDescriptor.getRoleManager().dropRole(AuthenticatedUser.SYSTEM_USER, RoleResource.role(USERNAME));
+            });
+        });
+    }
+
+    private void test(Consumer<IInvokableInstance> action) throws Exception
+    {
+        ICluster<IInvokableInstance> CLUSTER = cluster();
+        com.datastax.driver.core.Cluster driver = JavaDriverUtils.create(CLUSTER, null, builder -> builder
+                                                                                                   .withAuthProvider(CLIENT_AUTH_PROVIDER)
+                                                                                                   .withReconnectionPolicy(RECONNECTION_POLICY));
+        Session session = driver.connect();
+
+        // One control, one data connection per host
+        Assertions.assertThat(driver.getMetrics().getOpenConnections().getValue()).isEqualTo(CLUSTER_SIZE + 1);
+        Assertions.assertThat(driver.getMetrics().getConnectedToHosts().getValue()).isEqualTo(CLUSTER_SIZE);
+        Assertions.assertThat(driver.getMetrics().getErrorMetrics().getAuthenticationErrors().getCount()).isEqualTo(0);
+
+        action.accept(CLUSTER.get(1));
+
+        CLUSTER.forEach(instance -> {
+            instance.runOnInstance(() -> {
+                Roles.cache.invalidate();
+                StorageService.instance.disconnectInvalidRoles();
+            });
+        });
+
+        await().pollDelay(100, MILLISECONDS)
+               .pollInterval(100, MILLISECONDS)
+               .atMost(10, SECONDS)
+               .untilAsserted(() -> {
+                   // Should disconnect from both the coordinator of the DDL and the replica that is notified
+                   Assertions.assertThat(driver.getMetrics().getOpenConnections().getValue()).isEqualTo(0);
+                   Assertions.assertThat(driver.getMetrics().getConnectedToHosts().getValue()).isEqualTo(0);
+               });
+
+        await().pollDelay(100, MILLISECONDS)
+               .pollInterval(100, MILLISECONDS)
+               .atMost(10, SECONDS)
+               .untilAsserted(() -> {
+                   long authErrors = session.getCluster().getMetrics().getErrorMetrics().getAuthenticationErrors().getCount();
+                   Assertions.assertThat(authErrors).isGreaterThan(0);
+               });
+
+        session.close();
+        driver.close();
+        CLUSTER.close();
+    }
+}

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

@@ -31,6 +31,7 @@
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ConsistencyLevel;
+import org.assertj.core.api.Assertions;
 
 import static org.apache.cassandra.auth.AuthTestUtils.ALL_ROLES;
 import static org.apache.cassandra.auth.AuthTestUtils.ROLE_A;
@@ -139,4 +140,13 @@ public void testSuperUsers()
                                  .map(RoleResource::getRoleName)
                                  .collect(Collectors.toSet()));
     }
+
+    @Test
+    public void testNonexistentRoleCantLogin()
+    {
+        // There can be a reference to a nonexistent role (that has been removed from the cache and the system table)
+        // via the native transport connection state, make sure there's no NPE on canLogin check
+        AuthenticatedUser nonexistent = new AuthenticatedUser("nonexistent");
+        Assertions.assertThat(nonexistent.canLogin()).isFalse();
+    }
 }