HCD-23 Tighten up permissions on system keyspaces

CHANGES.txt

@@ -24,6 +24,10 @@ Merged from 4.0:
  * Fix a race condition where a keyspace can be oopened while it is being removed (CASSANDRA-17658)
 
 
+4.0.17
+ * Tighten up permission on system keyspaces (CASSANDRA-20040)
+
+
 4.0.12
  * Migrate Python optparse to argparse (CASSANDRA-17914)
 Merged from 3.11:

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

@@ -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)));
 }

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

@@ -19,26 +19,43 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Predicate;
 
 import org.apache.cassandra.utils.Hex;
 
 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;
     }

src/java/org/apache/cassandra/cql3/statements/GrantPermissionsStatement.java

@@ -17,20 +17,25 @@
  */
 package org.apache.cassandra.cql3.statements;
 
+import java.util.Collections;
 import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.cassandra.audit.AuditLogContext;
 import org.apache.cassandra.audit.AuditLogEntryType;
+import org.apache.cassandra.auth.DataResource;
 import org.apache.cassandra.auth.IAuthorizer;
 import org.apache.cassandra.auth.IResource;
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.config.DatabaseDescriptor;
 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.schema.SchemaConstants;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.ClientWarn;
+import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.transport.messages.ResultMessage;
 
 public class GrantPermissionsStatement extends PermissionsManagementStatement
@@ -40,6 +45,23 @@ public GrantPermissionsStatement(Set<Permission> permissions, IResource resource
         super(permissions, resource, grantee);
     }
 
+    public void validate(QueryState 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()
+                && SchemaConstants.isNonVirtualSystemKeyspace(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
     {
         IAuthorizer authorizer = DatabaseDescriptor.getAuthorizer();

src/java/org/apache/cassandra/schema/SchemaConstants.java

@@ -172,7 +172,17 @@ public static boolean isSystemKeyspace(String keyspaceName)
                 || isReplicatedSystemKeyspace(keyspaceName)
                 || isVirtualSystemKeyspace(keyspaceName);
     }
-
+
+    /**
+     * @return whether or not the keyspace is a non-virtual, system keyspace
+     */
+    public static boolean isNonVirtualSystemKeyspace(String keyspaceName)
+    {
+        final String lowercaseKeyspaceName = keyspaceName.toLowerCase();
+        return LOCAL_SYSTEM_KEYSPACE_NAMES.contains(lowercaseKeyspaceName)
+               || REPLICATED_SYSTEM_KEYSPACE_NAMES.contains(lowercaseKeyspaceName);
+    }
+
     /**
      * @return whether or not the keyspace is a virtual keyspace (system_virtual_schema, system_views)
      */

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

@@ -22,6 +22,7 @@
 import java.net.SocketAddress;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
@@ -403,9 +404,12 @@ private void ensurePermission(String keyspace, Permission perm, DataResource res
 
         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));
@@ -422,6 +426,24 @@ public void ensurePermission(Permission perm, IResource resource)
             if (((FunctionResource)resource).getKeyspace().equals(SchemaConstants.SYSTEM_KEYSPACE_NAME))
                 return;
 
+        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 (SchemaConstants.isSystemKeyspace(keyspace))
+                {
+                    ensurePermissionOnResourceChain(perm, Resources.chain(dataResource, IResource::hasParent));
+                    return;
+                }
+            }
+        }
+
         ensurePermissionOnResourceChain(perm, resource);
     }
 
@@ -444,7 +466,13 @@ public void ensurePermission(Permission permission, Function function)
 
     private void ensurePermissionOnResourceChain(Permission perm, IResource resource)
     {
-        for (IResource r : Resources.chain(resource))
+        ensurePermissionOnResourceChain(perm, Resources.chain(resource));
+    }
+
+    private void ensurePermissionOnResourceChain(Permission perm, List<? extends IResource> resources)
+    {
+        IResource resource = resources.get(0);
+        for (IResource r : resources)
             if (authorize(r).contains(perm))
                 return;