Apache Cassandra: User with MODIFY permission on ALL KEYSPACES can escalate privileges to superuser via unsafe actions

@@ -1,4 +1,5 @@
 4.0.16
+ * Tighten up permissions in dc authorizers (CASSANDRA-20225)
  * CBUtil serialization of UTF8 does not handle all UTF8 properly (CASSANDRA-20234)
  * Make hint expiry use request start time rather than timeout time for TTL (CASSANDRA-20014)
  * Do not attach rows and partitions to CheckForAbort when already attached (CASSANDRA-20135)

@@ -78,6 +78,9 @@ public void authorize(ClientState state) throws UnauthorizedException
         if (opts.getSuperuser().isPresent() && !isSuper)
             throw new UnauthorizedException("Only superusers are allowed to alter superuser status");
 
+        if (dcPermissions != null && !isSuper)
+            throw new UnauthorizedException("Only superusers are allowed to alter access to datacenters.");
+
         // superusers can do whatever else they like
         if (isSuper)
             return;

@@ -40,7 +40,9 @@
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.service.ClientState;
+import org.assertj.core.api.Assertions;
 
 import static org.apache.cassandra.auth.AuthKeyspace.NETWORK_PERMISSIONS;
 import static org.apache.cassandra.auth.RoleTestUtils.LocalCassandraRoleManager;
@@ -113,7 +115,19 @@ private static ClientState getClientState()
         return state;
     }
 
+    private static ClientState getClientState(String role)
+    {
+        ClientState state = ClientState.forInternalCalls();
+        state.login(new AuthenticatedUser(role));
+        return state;
+    }
+
     private static void auth(String query, Object... args)
+    {
+        auth(query, getClientState(), args);
+    }
+
+    private static void auth(String query, ClientState clientState, Object... args)
     {
         CQLStatement statement = QueryProcessor.parseStatement(String.format(query, args)).prepare(ClientState.forInternalCalls());
         assert statement instanceof CreateRoleStatement
@@ -123,7 +137,8 @@ private static void auth(String query, Object... args)
 
         // invalidate roles cache so that any changes to the underlying roles are picked up
         Roles.clearCache();
-        authStmt.execute(getClientState());
+        authStmt.authorize(clientState);
+        authStmt.execute(clientState);
     }
 
     private static DCPermissions dcPerms(String username)
@@ -170,6 +185,27 @@ public void alter()
         assertDcPermRow(username);
     }
 
+    @Test
+    public void alterAsUser()
+    {
+        String username = createName();
+
+        assertNoDcPermRow(username);
+        auth("CREATE ROLE %s WITH PASSWORD = 'password' AND LOGIN = true AND ACCESS TO DATACENTERS {'dc1'}", username);
+        Assert.assertEquals(DCPermissions.subset("dc1"), dcPerms(username));
+        assertDcPermRow(username, "dc1");
+
+        // try to alter as a user
+        ClientState userState = getClientState(username);
+        Assertions.assertThatThrownBy(() -> auth("ALTER ROLE %s WITH ACCESS TO DATACENTERS {'dc1', 'dc2'}", userState, username))
+        .hasMessage("Only superusers are allowed to alter access to datacenters.")
+        .isInstanceOf(UnauthorizedException.class);
+
+        // nothing changed
+        Assert.assertEquals(DCPermissions.subset("dc1"), dcPerms(username));
+        assertDcPermRow(username, "dc1");
+    }
+
     @Test
     public void drop()
     {

@@ -1,5 +1,6 @@
 3.0.31
- * Fix incorrect column identifier bytes problem when renaming a column (CASSANDRA-18965)
+ * Tighten up permissions on system keyspaces (CASSANDRA-20090)
+ * Fix incorrect column identifier bytes problem when renaming a column (CASSANDRA-18956)
  * Upgrade OWASP to 10.0.0 (CASSANDRA-19738)
 Merged from 2.2:
  * Add termin-8-jdk as a valid jdk8 candidate in the debian package (CASSANDRA-19752)

@@ -66,4 +66,11 @@ public enum Permission
     public static final Set<Permission> ALL =
             Sets.immutableEnumSet(EnumSet.range(Permission.CREATE, Permission.EXECUTE));
     public static final Set<Permission> NONE = ImmutableSet.of();
+
+    /**
+     * Set of Permissions which may never be granted on any system keyspace, or table in a system keyspace, to any role.
+     * (Only SELECT, DESCRIBE and ALTER may ever be granted).
+     */
+    public static final Set<Permission> INVALID_FOR_SYSTEM_KEYSPACES =
+            Sets.immutableEnumSet(EnumSet.complementOf(EnumSet.of(Permission.SELECT, Permission.DESCRIBE, Permission.ALTER)));
 }

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Predicate;
 
 import org.apache.cassandra.utils.Hex;
 
@@ -27,18 +28,33 @@ public final class Resources
     /**
      * Construct a chain of resource parents starting with the resource and ending with the root.
      *
-     * @param resource The staring point.
+     * @param resource The starting point.
      * @return list of resource in the chain form start to the root.
      */
     public static List<? extends IResource> chain(IResource resource)
     {
-        List<IResource> chain = new ArrayList<IResource>();
+        return chain(resource, (r) -> true);
+    }
+
+    /**
+     * Construct a chain of resource parents starting with the resource and ending with the root. Only resources which
+     * satisfy the supplied predicate will be included.
+     *
+     * @param resource The starting point.
+     * @param filter can be used to omit specific resources from the chain
+     * @return list of resource in the chain form start to the root.
+     */
+    public static List<? extends IResource> chain(IResource resource, Predicate<IResource> filter)
+    {
+
+        List<IResource> chain = new ArrayList<>(4);
         while (true)
         {
-           chain.add(resource);
-           if (!resource.hasParent())
-               break;
-           resource = resource.getParent();
+            if (filter.test(resource))
+                chain.add(resource);
+            if (!resource.hasParent())
+                break;
+            resource = resource.getParent();
         }
         return chain;
     }

@@ -862,16 +862,31 @@ public static IAuthenticator getAuthenticator()
         return authenticator;
     }
 
+    public static void setAuthenticator(IAuthenticator authenticator)
+    {
+        DatabaseDescriptor.authenticator = authenticator;
+    }
+
     public static IAuthorizer getAuthorizer()
     {
         return authorizer;
     }
 
+    public static void setAuthorizer(IAuthorizer authorizer)
+    {
+        DatabaseDescriptor.authorizer = authorizer;
+    }
+
     public static IRoleManager getRoleManager()
     {
         return roleManager;
     }
 
+    public static void setRoleManager(IRoleManager roleManager)
+    {
+        DatabaseDescriptor.roleManager = roleManager;
+    }
+
     public static int getPermissionsValidity()
     {
         return conf.permissions_validity_in_ms;

@@ -125,6 +125,12 @@ public static boolean isReplicatedSystemKeyspace(String keyspaceName)
         return REPLICATED_SYSTEM_KEYSPACE_NAMES.contains(keyspaceName.toLowerCase());
     }
 
+    public static boolean isSystemKeyspace(String keyspaceName)
+    {
+        final String lowercaseKeyspaceName = keyspaceName.toLowerCase();
+        return LOCAL_SYSTEM_KEYSPACE_NAMES.contains(lowercaseKeyspaceName)
+               || REPLICATED_SYSTEM_KEYSPACE_NAMES.contains(lowercaseKeyspaceName);
+    }
     /**
      * load keyspace (keyspace) definitions, but do not initialize the keyspace instances.
      * Schema version may be updated as the result.

@@ -17,14 +17,18 @@
  */
 package org.apache.cassandra.cql3.statements;
 
+import java.util.Collections;
 import java.util.Set;
 
+import org.apache.cassandra.auth.DataResource;
 import org.apache.cassandra.auth.IResource;
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.cql3.RoleName;
 import org.apache.cassandra.exceptions.RequestExecutionException;
 import org.apache.cassandra.exceptions.RequestValidationException;
+import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.transport.messages.ResultMessage;
 
@@ -35,6 +39,23 @@ public GrantPermissionsStatement(Set<Permission> permissions, IResource resource
         super(permissions, resource, grantee);
     }
 
+    public void validate(ClientState state) throws RequestValidationException
+    {
+        super.validate(state);
+        if (resource instanceof DataResource)
+        {
+            DataResource data = (DataResource) resource;
+            // Only a subset of permissions can be granted on system keyspaces
+            if (!data.isRootLevel()
+                && Schema.isSystemKeyspace(data.getKeyspace())
+                && !Collections.disjoint(permissions, Permission.INVALID_FOR_SYSTEM_KEYSPACES))
+            {
+                throw new UnauthorizedException("Granting permissions on system keyspaces is strictly limited, " +
+                                                "this operation is not permitted");
+            }
+        }
+    }
+
     public ResultMessage execute(ClientState state) throws RequestValidationException, RequestExecutionException
     {
         DatabaseDescriptor.getAuthorizer().grant(state.getUser(), permissions, resource, grantee);

@@ -20,10 +20,11 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Arrays;
-import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 
+import com.google.common.collect.ImmutableSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,6 +42,7 @@
 import org.apache.cassandra.exceptions.UnauthorizedException;
 import org.apache.cassandra.schema.SchemaKeyspace;
 import org.apache.cassandra.thrift.ThriftValidation;
+import org.apache.cassandra.tracing.TraceKeyspace;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.CassandraVersion;
@@ -54,29 +56,40 @@ public class ClientState
     private static final Logger logger = LoggerFactory.getLogger(ClientState.class);
     public static final CassandraVersion DEFAULT_CQL_VERSION = org.apache.cassandra.cql3.QueryProcessor.CQL_VERSION;
 
-    private static final Set<IResource> READABLE_SYSTEM_RESOURCES = new HashSet<>();
-    private static final Set<IResource> PROTECTED_AUTH_RESOURCES = new HashSet<>();
-    private static final Set<IResource> DROPPABLE_SYSTEM_AUTH_TABLES = new HashSet<>();
+    public static final ImmutableSet<IResource> READABLE_SYSTEM_RESOURCES;
+    public static final ImmutableSet<IResource> PROTECTED_AUTH_RESOURCES;
+    public static final ImmutableSet<IResource> DROPPABLE_SYSTEM_AUTH_TABLES;
     static
     {
         // We want these system cfs to be always readable to authenticated users since many tools rely on them
         // (nodetool, cqlsh, bulkloader, etc.)
-        for (String cf : Arrays.asList(SystemKeyspace.LOCAL, SystemKeyspace.PEERS))
-            READABLE_SYSTEM_RESOURCES.add(DataResource.table(SystemKeyspace.NAME, cf));
+        ImmutableSet.Builder<IResource> readableBuilder = ImmutableSet.builder();
+        for (String cf : Arrays.asList(SystemKeyspace.LOCAL, SystemKeyspace.PEERS, SystemKeyspace.SIZE_ESTIMATES))
+            readableBuilder.add(DataResource.table(SystemKeyspace.NAME, cf));
 
-        SchemaKeyspace.ALL.forEach(table -> READABLE_SYSTEM_RESOURCES.add(DataResource.table(SchemaKeyspace.NAME, table)));
+        // make all schema tables readable by default (required by the drivers)
+        SchemaKeyspace.ALL.forEach(table -> readableBuilder.add(DataResource.table(SchemaKeyspace.NAME, table)));
 
+        // make system_traces readable by all or else tracing will require explicit grants
+        readableBuilder.add(DataResource.table(TraceKeyspace.NAME, TraceKeyspace.EVENTS));
+        readableBuilder.add(DataResource.table(TraceKeyspace.NAME, TraceKeyspace.SESSIONS));
+        READABLE_SYSTEM_RESOURCES = readableBuilder.build();
+
+        ImmutableSet.Builder<IResource> protectedBuilder = ImmutableSet.builder();
+        // neither clients nor tools need authentication/authorization
         if (!Config.isClientMode())
         {
-            PROTECTED_AUTH_RESOURCES.addAll(DatabaseDescriptor.getAuthenticator().protectedResources());
-            PROTECTED_AUTH_RESOURCES.addAll(DatabaseDescriptor.getAuthorizer().protectedResources());
-            PROTECTED_AUTH_RESOURCES.addAll(DatabaseDescriptor.getRoleManager().protectedResources());
+            protectedBuilder.addAll(DatabaseDescriptor.getAuthenticator().protectedResources());
+            protectedBuilder.addAll(DatabaseDescriptor.getAuthorizer().protectedResources());
+            protectedBuilder.addAll(DatabaseDescriptor.getRoleManager().protectedResources());
         }
-
+        PROTECTED_AUTH_RESOURCES = protectedBuilder.build();
+        ImmutableSet.Builder<IResource> droppableBuilder = ImmutableSet.builder();
         // allow users with sufficient privileges to drop legacy tables (users, credentials, permissions) from AUTH_KS
-        DROPPABLE_SYSTEM_AUTH_TABLES.add(DataResource.table(AuthKeyspace.NAME, PasswordAuthenticator.LEGACY_CREDENTIALS_TABLE));
-        DROPPABLE_SYSTEM_AUTH_TABLES.add(DataResource.table(AuthKeyspace.NAME, CassandraRoleManager.LEGACY_USERS_TABLE));
-        DROPPABLE_SYSTEM_AUTH_TABLES.add(DataResource.table(AuthKeyspace.NAME, CassandraAuthorizer.USER_PERMISSIONS));
+        droppableBuilder.add(DataResource.table(AuthKeyspace.NAME, PasswordAuthenticator.LEGACY_CREDENTIALS_TABLE));
+        droppableBuilder.add(DataResource.table(AuthKeyspace.NAME, CassandraRoleManager.LEGACY_USERS_TABLE));
+        droppableBuilder.add(DataResource.table(AuthKeyspace.NAME, CassandraAuthorizer.USER_PERMISSIONS));
+        DROPPABLE_SYSTEM_AUTH_TABLES = droppableBuilder.build();
     }
 
     // Current user for the session
@@ -328,9 +341,12 @@ private void hasAccess(String keyspace, Permission perm, DataResource resource)
 
         preventSystemKSSchemaModification(keyspace, resource, perm);
 
+        // Some system data is always readable
         if ((perm == Permission.SELECT) && READABLE_SYSTEM_RESOURCES.contains(resource))
             return;
 
+        // Modifications to any resource upon which the authenticator, authorizer or role manager depend should not be
+        // be performed by users
         if (PROTECTED_AUTH_RESOURCES.contains(resource))
             if ((perm == Permission.CREATE) || (perm == Permission.ALTER) || (perm == Permission.DROP))
                 throw new UnauthorizedException(String.format("%s schema is protected", resource));
@@ -348,7 +364,25 @@ public void ensureHasPermission(Permission perm, IResource resource) throws Unau
             if (((FunctionResource)resource).getKeyspace().equals(SystemKeyspace.NAME))
                 return;
 
-        checkPermissionOnResourceChain(perm, resource);
+        if (resource instanceof DataResource && !(user.isSuper() || user.isSystem()))
+        {
+            DataResource dataResource = (DataResource)resource;
+            if (!dataResource.isRootLevel())
+            {
+                String keyspace = dataResource.getKeyspace();
+                // A user may have permissions granted on ALL KEYSPACES, but this should exclude system keyspaces. Any
+                // permission on those keyspaces or their tables must be granted to the user either explicitly or
+                // transitively. The set of grantable permissions for system keyspaces is further limited,
+                // see the Permission enum for details.
+                if (Schema.isSystemKeyspace(keyspace))
+                {
+                    ensurePermissionOnResourceChain(perm, Resources.chain(dataResource, IResource::hasParent));
+                    return;
+                }
+            }
+        }
+
+        ensurePermissionOnResourceChain(perm, resource);
     }
 
     // Convenience method called from checkAccess method of CQLStatement
@@ -363,14 +397,20 @@ public void ensureHasPermission(Permission permission, Function function)
         if (function.isNative())
             return;
 
-        checkPermissionOnResourceChain(permission, FunctionResource.function(function.name().keyspace,
-                                                                             function.name().name,
-                                                                             function.argTypes()));
+        ensurePermissionOnResourceChain(permission, FunctionResource.function(function.name().keyspace,
+                                                                              function.name().name,
+                                                                              function.argTypes()));
+    }
+
+    private void ensurePermissionOnResourceChain(Permission perm, IResource resource)
+    {
+        ensurePermissionOnResourceChain(perm, Resources.chain(resource));
     }
 
-    private void checkPermissionOnResourceChain(Permission perm, IResource resource)
+    private void ensurePermissionOnResourceChain(Permission perm, List<? extends IResource> resources)
     {
-        for (IResource r : Resources.chain(resource))
+        IResource resource = resources.get(0);
+        for (IResource r : resources)
             if (authorize(r).contains(perm))
                 return;
 

@@ -0,0 +1,375 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import com.google.common.collect.Iterables;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.PlainTextAuthProvider;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.exceptions.UnauthorizedException;
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.metrics.ClientMetrics;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.SchemaKeyspace;
+import org.apache.cassandra.schema.Types;
+import org.apache.cassandra.tracing.TraceKeyspace;
+import org.apache.cassandra.transport.ConfiguredLimit;
+import org.apache.cassandra.transport.Server;
+import org.apache.cassandra.utils.Pair;
+
+import static java.lang.String.format;
+import static org.apache.cassandra.config.Schema.LOCAL_SYSTEM_KEYSPACE_NAMES;
+import static org.apache.cassandra.config.Schema.REPLICATED_SYSTEM_KEYSPACE_NAMES;
+
+public class GrantAndRevokeTest extends CQLTester
+{
+    private static final String user = "user";
+    private static final String pass = "12345";
+    private static final Pair<String, String> USER = Pair.create(user, pass);
+    private static final Pair<String, String> SUPERUSER = Pair.create("cassandra", "cassandra");
+    private static int counter = 0;
+
+    @BeforeClass
+    public static void setUpClass()
+    {
+        System.setProperty("cassandra.superuser_setup_delay_ms", "0");
+        DatabaseDescriptor.setPermissionsValidity(0);
+        CQLTester.setUpClass();
+        SchemaLoader.createKeyspace(AuthKeyspace.NAME, KeyspaceParams.simple(1), AuthKeyspace.metadata().tables, Types.none());
+
+        DatabaseDescriptor.setAuthenticator(new PasswordAuthenticator());
+        DatabaseDescriptor.setRoleManager(new CassandraRoleManager());
+        DatabaseDescriptor.setAuthorizer(new CassandraAuthorizer());
+        // needed as the driver reads system.local to get the release version, which determines the
+        // schema tables it attempts to read (current, or legacy in case it's connecting to a 2.2 node)
+        SystemKeyspace.persistLocalMetadata();
+        prepareNetwork();
+        protocolVersionLimit = ConfiguredLimit.newLimit();
+        server = new Server.Builder().withHost(nativeAddr)
+                                     .withPort(nativePort)
+                                     .withProtocolVersionLimit(protocolVersionLimit)
+                                     .build();
+        ClientMetrics.instance.init(Collections.singleton(server));
+        server.start();
+    }
+
+    @After
+    public void tearDown() throws Throwable
+    {
+        session(SUPERUSER).execute("DROP ROLE " + user);
+    }
+
+    private Session session(Pair<String, String> credentials)
+    {
+        Cluster cluster = Cluster.builder()
+                                 .addContactPoints(nativeAddr)
+                                 .withClusterName("Test Cluster " + counter++)
+                                 .withPort(nativePort)
+                                 .withAuthProvider(new PlainTextAuthProvider(credentials.left, credentials.right))
+                                 .build();
+
+        return cluster.connect();
+    }
+
+    @Test
+    public void testGrantedKeyspace() throws Throwable
+    {
+        Session superuser = session(SUPERUSER);
+        superuser.execute(String.format("CREATE ROLE %s WITH LOGIN = TRUE AND password='%s'", user, pass));
+        superuser.execute("GRANT CREATE ON KEYSPACE " + KEYSPACE_PER_TEST + " TO " + user);
+        String table = KEYSPACE_PER_TEST + '.' + createTable(KEYSPACE_PER_TEST, "CREATE TABLE %s (pk int, ck int, val int, val_2 text, PRIMARY KEY (pk, ck))");
+        String index = KEYSPACE_PER_TEST + ".idx_01";
+        createIndex(KEYSPACE_PER_TEST, "CREATE INDEX idx_01 ON %s (val_2)");
+        String type = KEYSPACE_PER_TEST + '.' + createType(KEYSPACE_PER_TEST, "CREATE TYPE %s (a int, b text)");
+        String mv = KEYSPACE_PER_TEST + ".ks_mv_01";
+        superuser.execute("CREATE MATERIALIZED VIEW " + mv + " AS SELECT * FROM " + table + " WHERE val IS NOT NULL AND pk IS NOT NULL AND ck IS NOT NULL PRIMARY KEY (val, pk, ck)");
+
+        Session nonsuperuser = session(USER);
+
+        // ALTER and DROP tables created by somebody else
+        // Spin assert for effective auth changes.
+        final String spinAssertTable = table;
+        Util.spinAssertEquals(false, () -> {
+            try
+            {
+                assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + spinAssertTable + "> or any of its parents",
+                                        formatQuery(KEYSPACE_PER_TEST, "INSERT INTO %s (pk, ck, val, val_2) VALUES (1, 1, 1, '1')"));
+            }
+            catch(Throwable e)
+            {
+                return true;
+            }
+            return false;
+        }, 10);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "UPDATE %s SET val = 1 WHERE pk = 1 AND ck = 1"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "DELETE FROM %s WHERE pk = 1 AND ck = 2"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no SELECT permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "SELECT * FROM %s WHERE pk = 1 AND ck = 1"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no SELECT permission on <table " + table + "> or any of its parents",
+                                "SELECT * FROM " + mv + " WHERE val = 1 AND pk = 1 AND ck = 1");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "TRUNCATE TABLE %s"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "ALTER TABLE %s ADD val_3 int"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no DROP permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "DROP TABLE %s"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <keyspace " + KEYSPACE_PER_TEST + "> or any of its parents",
+                                "ALTER TYPE " + type + " ADD c bigint");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no DROP permission on <keyspace " + KEYSPACE_PER_TEST + "> or any of its parents",
+                                "DROP TYPE " + type);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <table " + table + "> or any of its parents",
+                                "DROP MATERIALIZED VIEW " + mv);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <table " + table + "> or any of its parents",
+                                "DROP INDEX " + index);
+
+        superuser.execute("GRANT ALTER ON KEYSPACE " + KEYSPACE_PER_TEST + " TO " + user);
+        superuser.execute("GRANT DROP ON KEYSPACE " + KEYSPACE_PER_TEST + " TO " + user);
+        superuser.execute("GRANT SELECT ON KEYSPACE " + KEYSPACE_PER_TEST + " TO " + user);
+        superuser.execute("GRANT MODIFY ON KEYSPACE " + KEYSPACE_PER_TEST + " TO " + user);
+        // Spin assert for effective auth changes.
+        Util.spinAssertEquals(false, () -> {
+            try
+            {
+                nonsuperuser.execute("ALTER KEYSPACE " + KEYSPACE_PER_TEST + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}");
+            }
+            catch(Throwable e)
+            {
+                return true;
+            }
+            return false;
+        }, 10);
+
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "INSERT INTO %s (pk, ck, val, val_2) VALUES (1, 1, 1, '1')"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "UPDATE %s SET val = 1 WHERE pk = 1 AND ck = 1"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "DELETE FROM %s WHERE pk = 1 AND ck = 2"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "SELECT * FROM %s WHERE pk = 1 AND ck = 1"));
+        nonsuperuser.execute("SELECT * FROM " + mv + " WHERE val = 1 AND pk = 1");
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "TRUNCATE TABLE %s"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "ALTER TABLE %s ADD val_3 int"));
+        nonsuperuser.execute("DROP MATERIALIZED VIEW " + mv);
+        nonsuperuser.execute("DROP INDEX " + index);
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "DROP TABLE %s"));
+        nonsuperuser.execute("ALTER TYPE " + type + " ADD c bigint");
+        nonsuperuser.execute("DROP TYPE " + type);
+
+        // calling creatTableName to create a new table name that will be used by the formatQuery
+        table = createTableName();
+        type = KEYSPACE_PER_TEST + "." + createTypeName();
+        mv = KEYSPACE_PER_TEST + ".ks_mv_02";
+        nonsuperuser.execute("CREATE TYPE " + type + " (a int, b text)");
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "CREATE TABLE %s (pk int, ck int, val int, val_2 text, PRIMARY KEY (pk, ck))"));
+        nonsuperuser.execute("CREATE MATERIALIZED VIEW " + mv + " AS SELECT * FROM " + table + " WHERE val IS NOT NULL AND pk IS NOT NULL AND ck IS NOT NULL PRIMARY KEY (val, pk, ck)");
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "INSERT INTO %s (pk, ck, val, val_2) VALUES (1, 1, 1, '1')"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "UPDATE %s SET val = 1 WHERE pk = 1 AND ck = 1"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "DELETE FROM %s WHERE pk = 1 AND ck = 2"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "SELECT * FROM %s WHERE pk = 1 AND ck = 1"));
+        nonsuperuser.execute("SELECT * FROM " + mv + " WHERE val = 1 AND pk = 1");
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "TRUNCATE TABLE %s"));
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "ALTER TABLE %s ADD val_3 int"));
+        nonsuperuser.execute("DROP MATERIALIZED VIEW " + mv);
+        nonsuperuser.execute(formatQuery(KEYSPACE_PER_TEST, "DROP TABLE %s"));
+        nonsuperuser.execute("ALTER TYPE " + type + " ADD c bigint");
+        nonsuperuser.execute("DROP TYPE " + type);
+
+        superuser.execute("REVOKE ALTER ON KEYSPACE " + KEYSPACE_PER_TEST + " FROM " + user);
+        superuser.execute("REVOKE DROP ON KEYSPACE " + KEYSPACE_PER_TEST + " FROM " + user);
+        superuser.execute("REVOKE SELECT ON KEYSPACE " + KEYSPACE_PER_TEST + " FROM " + user);
+        superuser.execute("REVOKE MODIFY ON KEYSPACE " + KEYSPACE_PER_TEST + " FROM " + user);
+
+        table = KEYSPACE_PER_TEST + "." + createTable(KEYSPACE_PER_TEST, "CREATE TABLE %s (pk int, ck int, val int, val_2 text, PRIMARY KEY (pk, ck))");
+        type = KEYSPACE_PER_TEST + "." + createType(KEYSPACE_PER_TEST, "CREATE TYPE %s (a int, b text)");
+        index = KEYSPACE_PER_TEST + ".idx_02";
+        createIndex(KEYSPACE_PER_TEST, "CREATE INDEX idx_02 ON %s (val_2)");
+        mv = KEYSPACE_PER_TEST + ".ks_mv_03";
+        superuser.execute("CREATE MATERIALIZED VIEW " + mv + " AS SELECT * FROM " + table + " WHERE val IS NOT NULL AND pk IS NOT NULL AND ck IS NOT NULL PRIMARY KEY (val, pk, ck)");
+
+        // Spin assert for effective auth changes.
+        final String spinAssertTable2 = table;
+        Util.spinAssertEquals(false, () -> {
+            try
+            {
+                assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + spinAssertTable2 + "> or any of its parents",
+                                        "INSERT INTO " + spinAssertTable2 + " (pk, ck, val, val_2) VALUES (1, 1, 1, '1')");
+            }
+            catch(Throwable e)
+            {
+                return true;
+            }
+            return false;
+        }, 10);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + table + "> or any of its parents",
+                                "UPDATE " + table + " SET val = 1 WHERE pk = 1 AND ck = 1");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + table + "> or any of its parents",
+                                "DELETE FROM " + table + " WHERE pk = 1 AND ck = 2");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no SELECT permission on <table " + table + "> or any of its parents",
+                                "SELECT * FROM " + table + " WHERE pk = 1 AND ck = 1");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no SELECT permission on <table " + table + "> or any of its parents",
+                                "SELECT * FROM " + mv + " WHERE val = 1 AND pk = 1 AND ck = 1");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table " + table + "> or any of its parents",
+                                "TRUNCATE TABLE " + table);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "ALTER TABLE " + table + " ADD val_3 int"));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no DROP permission on <table " + table + "> or any of its parents",
+                                formatQuery(KEYSPACE_PER_TEST, "DROP TABLE " + table));
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <keyspace " + KEYSPACE_PER_TEST + "> or any of its parents",
+                                "ALTER TYPE " + type + " ADD c bigint");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no DROP permission on <keyspace " + KEYSPACE_PER_TEST + "> or any of its parents",
+                                "DROP TYPE " + type);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <table " + table + "> or any of its parents",
+                                "DROP MATERIALIZED VIEW " + mv);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no ALTER permission on <table " + table + "> or any of its parents",
+                                "DROP INDEX " + index);
+    }
+
+
+    @Test
+    public void testSpecificGrantsOnSystemKeyspaces() throws Throwable
+    {
+        // Granting specific permissions on system keyspaces should not be allowed if those permissions include any from
+        // the denylist Permission.INVALID_FOR_SYSTEM_KEYSPACES. By this definition, GRANT ALL on any system keyspace,
+        // or a table within one, should be rejected.
+        Session superuser = session(SUPERUSER);
+        superuser.execute("CREATE ROLE '" + user + "'");
+        String responseMsg = "Granting permissions on system keyspaces is strictly limited, this operation is not permitted";
+        for (String keyspace : Iterables.concat(LOCAL_SYSTEM_KEYSPACE_NAMES, REPLICATED_SYSTEM_KEYSPACE_NAMES))
+        {
+            assertUnauthorizedQuery(superuser, responseMsg, format("GRANT ALL PERMISSIONS ON KEYSPACE %s TO %s", keyspace, user));
+            DataResource keyspaceResource = DataResource.keyspace(keyspace);
+            for (Permission p : keyspaceResource.applicablePermissions())
+                maybeRejectGrant(superuser, p, responseMsg, format("GRANT %s ON KEYSPACE %s TO %s", p.name(), keyspace, user));
+
+            for (CFMetaData table : Schema.instance.getKSMetaData(keyspace).tables)
+            {
+                DataResource tableResource = DataResource.table(keyspace, table.cfName);
+                assertUnauthorizedQuery(superuser, responseMsg, format("GRANT ALL PERMISSIONS ON %s.\"%s\" TO %s", table.ksName, table.cfName, user));
+                for (Permission p : tableResource.applicablePermissions())
+                    maybeRejectGrant(superuser, p, responseMsg, format("GRANT %s ON %s.\"%s\" TO %s", p.name(), table.ksName, table.cfName, user));
+            }
+        }
+    }
+
+    @Test
+    public void testGrantOnAllKeyspaces() throws Throwable
+    {
+        // Granting either specific or ALL permissions on ALL KEYSPACES is allowed, however these permissions are
+        // effective for non-system keyspaces only. If for any reason it is necessary to modify permissions on
+        // on a system keyspace, it must be done using keyspace specific grant statements.
+        Session superuser = session(SUPERUSER);
+        superuser.execute(String.format("CREATE ROLE %s WITH LOGIN = TRUE AND password='%s'", user, pass));
+        superuser.execute(String.format("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}", TraceKeyspace.NAME));
+        superuser.execute("CREATE KEYSPACE user_keyspace WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}");
+        superuser.execute("CREATE TABLE user_keyspace.t1 (k int PRIMARY KEY)");
+
+        Session nonsuperuser = session(USER);
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table user_keyspace.t1> or any of its parents",
+                                "INSERT INTO user_keyspace.t1 (k) VALUES (0)");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table system.local> or any of its parents",
+                                "INSERT INTO system.local(key) VALUES ('invalid')");
+
+        superuser.execute(format("GRANT MODIFY ON ALL KEYSPACES TO %s", user));
+        // User now has write permission on non-system keyspaces only
+        nonsuperuser.execute("INSERT INTO user_keyspace.t1 (k) VALUES (0)");
+        assertUnauthorizedQuery(nonsuperuser, "User user has no MODIFY permission on <table system.local> or any of its parents",
+                                "INSERT INTO system.local(key) VALUES ('invalid')");
+
+        // A non-superuser only has read access to a pre-defined set of system tables and all system_schema/traces
+        // tables and granting ALL permissions on ALL keyspaces also does not affect this.
+        maybeReadSystemTables(nonsuperuser, false);
+        superuser.execute(format("GRANT ALL PERMISSIONS ON ALL KEYSPACES TO %s", user));
+        maybeReadSystemTables(nonsuperuser, false);
+
+        // A superuser can still read system tables
+        maybeReadSystemTables(superuser, true);
+        // and also write to them, though this is still strongly discouraged
+        superuser.execute("INSERT INTO system.peers(peer, data_center) VALUES ('127.0.100.100', 'invalid_dc')");
+
+    }
+
+    private void maybeReadSystemTables(Session session, boolean isSuper) throws Throwable
+    {
+        Set<String> readableKeyspaces = new HashSet<>(Arrays.asList(SchemaKeyspace.NAME, TraceKeyspace.NAME));
+        Set<String> readableSystemTables = new HashSet<>(Arrays.asList(SystemKeyspace.LOCAL,
+                                                                       SystemKeyspace.PEERS,
+                                                                       SystemKeyspace.SIZE_ESTIMATES));
+
+        for (String keyspace : Iterables.concat(LOCAL_SYSTEM_KEYSPACE_NAMES, REPLICATED_SYSTEM_KEYSPACE_NAMES))
+        {
+            for (CFMetaData table : Schema.instance.getKSMetaData(keyspace).tables)
+            {
+                if (isSuper || (readableKeyspaces.contains(keyspace) || (keyspace.equals(SystemKeyspace.NAME) && readableSystemTables.contains(table.cfName))))
+                {
+                    session.execute(format("SELECT * FROM %s.\"%s\" LIMIT 1", table.ksName, table.cfName));
+                }
+                else
+                {
+                    assertUnauthorizedQuery(session, format("User %s has no SELECT permission on %s or any of its parents", user, DataResource.table(table.ksName, table.cfName)),
+                                            format("SELECT * FROM %s.\"%s\" LIMIT 1", table.ksName, table.cfName));
+                }
+            }
+        }
+    }
+
+    private void maybeRejectGrant(Session session, Permission p, String errorResponse, String grant) throws Throwable
+    {
+        if (Permission.INVALID_FOR_SYSTEM_KEYSPACES.contains(p))
+            assertUnauthorizedQuery(session, errorResponse, grant);
+        else
+            session.execute(grant);
+    }
+
+    private void assertUnauthorizedQuery(Session session, String errorMessage, String query) throws Throwable
+    {
+        try
+        {
+            session.execute(query);
+            Assert.fail("Query should be invalid but no error was thrown. Query is: " + query);
+        }
+        catch (Exception e)
+        {
+            if (!UnauthorizedException.class.isAssignableFrom(e.getClass()))
+            {
+                Assert.fail("Query should be invalid but wrong error was thrown. " +
+                            "Expected: " + UnauthorizedException.class.getName() + ", got: " + e.getClass().getName() + ". " +
+                            "Query is: " + query);
+            }
+            if (errorMessage != null)
+            {
+                assertMessageContains(errorMessage, e);
+            }
+        }
+    }
+
+}

@@ -45,6 +45,7 @@
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
 import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.metrics.ClientMetrics;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.Schema;
@@ -87,7 +88,7 @@ public abstract class CQLTester
     private static final AtomicInteger seqNumber = new AtomicInteger();
     protected static final ByteBuffer TOO_BIG = ByteBuffer.allocate(FBUtilities.MAX_UNSIGNED_SHORT + 1024);
 
-    private static org.apache.cassandra.transport.Server server;
+    protected static org.apache.cassandra.transport.Server server;
     protected static final int nativePort;
     protected static final InetAddress nativeAddr;
     protected static ConfiguredLimit protocolVersionLimit;
@@ -567,9 +568,14 @@ protected void disablePreparedReuseForTest()
     }
 
     protected String createType(String query)
+    {
+        return createType(KEYSPACE, query);
+    }
+
+    protected String createType(String keyspace, String query)
     {
         String typeName = createTypeName();
-        String fullQuery = String.format(query, KEYSPACE + "." + typeName);
+        String fullQuery = String.format(query, keyspace + "." + typeName);
         logger.info(fullQuery);
         schemaChange(fullQuery);
         return typeName;
@@ -706,7 +712,12 @@ protected void dropFormattedTable(String formattedQuery)
 
     protected void createIndex(String query)
     {
-        createFormattedIndex(formatQuery(query));
+        createIndex(KEYSPACE, query);
+    }
+
+    protected void createIndex(String keyspace, String query)
+    {
+        createFormattedIndex(formatQuery(keyspace, query));
     }
 
     protected void createFormattedIndex(String formattedQuery)
@@ -803,10 +814,18 @@ protected CFMetaData currentTableMetadata()
         return Schema.instance.getCFMetaData(KEYSPACE, currentTable());
     }
 
+    protected com.datastax.driver.core.ResultSet executeNet(int protocolVersion, ConsistencyLevel consistency, String query) throws Throwable
+    {
+        Statement statement = new SimpleStatement(formatQuery(query));
+        statement = statement.setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel.valueOf(consistency.name()));
+        return sessionNet(protocolVersion).execute(statement);
+    }
+
     protected com.datastax.driver.core.ResultSet executeNet(String query, Object... values) throws Throwable
     {
         return sessionNet().execute(formatQuery(query), values);
     }
+
     protected com.datastax.driver.core.ResultSet executeNet(int protocolVersion, String query, Object... values) throws Throwable
     {
         return sessionNet(protocolVersion).execute(formatQuery(query), values);
@@ -1312,7 +1331,7 @@ protected void assertInvalidSyntaxMessage(String errorMessage, String query, Obj
      * @param text the text that the exception message must contains
      * @param e the exception to check
      */
-    private static void assertMessageContains(String text, Exception e)
+    protected static void assertMessageContains(String text, Exception e)
     {
         Assert.assertTrue("Expected error message to contain '" + text + "', but got '" + e.getMessage() + "'",
                 e.getMessage().contains(text));