[CASSANDRA-21606][trunk]Optimize BatchStatement.authorize for the case when a single table is updated

NEWS.txt

@@ -95,6 +95,11 @@ Upgrading
 ---------
 
     - trickle_fsync is by default set to "true" instead of "false" in cassandra.yaml (see CASSANDRA-21572)
+    - Authorizing a batch or an Accord transaction now performs the table permission checks once per run of
+      consecutive statements on the same table rather than once per statement. The permissions required are
+      unchanged, and the datacenter and CIDR checks that ride along with them still run at least once per
+      request. Auth cache request/hit/miss counts and the system_views.cidr_filtering_metrics_* counts and
+      latencies will therefore report fewer events than before. (see CASSANDRA-21606)
 
 Deprecation
 -----------

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

@@ -243,8 +243,9 @@ public boolean eligibleAsPreparedStatement()
     @Override
     public void authorize(ClientState state) throws InvalidRequestException, UnauthorizedException
     {
+        ModificationStatement.TableAuthorizationState authorizationState = new ModificationStatement.TableAuthorizationState();
         for (ModificationStatement statement : statements)
-            statement.authorize(state);
+            statement.authorize(state, authorizationState);
     }
 
     // Validates a prepared batch statement without validating its nested statements.

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

@@ -114,6 +114,7 @@
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.ViewMetadata;
 import org.apache.cassandra.service.ClientState;
@@ -380,28 +381,60 @@ public int getTimeToLive(FunctionContext context) throws InvalidRequestException
     @Override
     public void authorize(ClientState state) throws InvalidRequestException, UnauthorizedException
     {
-        state.ensureTablePermission(metadata, Permission.MODIFY);
+        authorize(state, null);
+    }
+
+    void authorize(ClientState state, TableAuthorizationState tableAuthorizationState) throws InvalidRequestException, UnauthorizedException
+    {
+        boolean tableChanged = tableAuthorizationState == null || !metadata.id.equals(tableAuthorizationState.tableId);
+        if (tableChanged)
+        {
+            state.ensureTablePermission(metadata, Permission.MODIFY);
+
+            if (tableAuthorizationState != null)
+            {
+                tableAuthorizationState.tableId = metadata.id;
+                tableAuthorizationState.selectAuthorized = false;
+            }
+        }
 
         // CAS updates can be used to simulate a SELECT query, so should require Permission.SELECT as well.
         if (hasConditions())
-            state.ensureTablePermission(metadata, Permission.SELECT);
+            ensureSelectPermission(state, tableAuthorizationState);
 
-        // MV updates need to get the current state from the table, and might update the views
-        // Require Permission.SELECT on the base table, and Permission.MODIFY on the views
-        Iterator<ViewMetadata> views = View.findAll(keyspace(), table()).iterator();
-        if (views.hasNext())
+        if (tableChanged)
         {
-            state.ensureTablePermission(metadata, Permission.SELECT);
-            do
+            // MV updates need to get the current state from the table, and might update the views
+            // Require Permission.SELECT on the base table, and Permission.MODIFY on the views
+            Iterator<ViewMetadata> views = View.findAll(keyspace(), table()).iterator();
+            if (views.hasNext())
             {
-                state.ensureTablePermission(views.next().metadata, Permission.MODIFY);
-            } while (views.hasNext());
+                ensureSelectPermission(state, tableAuthorizationState);
+                do
+                {
+                    state.ensureTablePermission(views.next().metadata, Permission.MODIFY);
+                } while (views.hasNext());
+            }
         }
 
         for (Function function : getFunctions())
             state.ensurePermission(Permission.EXECUTE, function);
     }
 
+    private void ensureSelectPermission(ClientState state, TableAuthorizationState tableAuthorizationState)
+    {
+        if (tableAuthorizationState == null || !tableAuthorizationState.selectAuthorized)
+            state.ensureTablePermission(metadata, Permission.SELECT);
+        if (tableAuthorizationState != null)
+            tableAuthorizationState.selectAuthorized = true;
+    }
+
+    static class TableAuthorizationState
+    {
+        private TableId tableId;
+        private boolean selectAuthorized;
+    }
+
     public void validate(ClientState state) throws InvalidRequestException
     {
         checkFalse(hasConditions() && attrs.isTimestampSet(), "Cannot provide custom timestamp for conditional updates");

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

@@ -233,8 +233,9 @@ public void authorize(ClientState state)
         if (returningSelect != null)
             returningSelect.select.authorize(state);
 
+        ModificationStatement.TableAuthorizationState authorizationState = new ModificationStatement.TableAuthorizationState();
         for (ModificationStatement update : updates)
-            update.authorize(state);
+            update.authorize(state, authorizationState);
     }
 
     @Override

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

@@ -0,0 +1,238 @@
+/*
+ * 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.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.cql3.Attributes;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.VariableSpecifications;
+import org.apache.cassandra.cql3.statements.BatchStatement;
+import org.apache.cassandra.cql3.statements.ModificationStatement;
+import org.apache.cassandra.exceptions.UnauthorizedException;
+import org.apache.cassandra.service.ClientState;
+
+import static org.apache.cassandra.auth.AuthTestUtils.auth;
+
+public class BatchAuthTest extends CQLTester
+{
+    private String table1;
+    private String table2;
+
+    @BeforeClass
+    public static void setUpAuth()
+    {
+        IRoleManager roleManager = new AuthTestUtils.LocalCassandraRoleManager();
+        SchemaLoader.setupAuth(roleManager,
+                               new AuthTestUtils.LocalPasswordAuthenticator(),
+                               new AuthTestUtils.LocalCassandraAuthorizer(),
+                               new AuthTestUtils.LocalCassandraNetworkAuthorizer(),
+                               new AuthTestUtils.LocalCassandraCIDRAuthorizer());
+        roleManager.setup();
+        AuthCacheService.initializeAndRegisterCaches();
+        AuthTestUtils.setupSuperUser();
+
+        requireNetwork();
+    }
+
+    @Before
+    public void setUpTest()
+    {
+        table1 = createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+        table2 = createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+    }
+
+    /**
+     * All statements target the same table
+     */
+    @Test
+    public void singleTableBatchRequiresModify()
+    {
+        ClientState clientState = createUserAndLogin();
+        BatchStatement batch = batch(clientState,
+                                     insert(table1, 0),
+                                     insert(table1, 1),
+                                     insert(table1, 2));
+
+        assertUnauthorized(batch, clientState, Permission.MODIFY, table1);
+
+        grant(clientState, Permission.MODIFY, table1);
+        batch.authorize(clientState);
+    }
+
+    /**
+     * Every table in the batch needs its own MODIFY check: state retained for one table must not satisfy the
+     * next one.
+     */
+    @Test
+    public void multiTableBatchRequiresModifyOnEveryTable()
+    {
+        ClientState clientState = createUserAndLogin();
+        BatchStatement batch = batch(clientState,
+                                     insert(table1, 0),
+                                     insert(table2, 0),
+                                     insert(table1, 1));
+
+        assertUnauthorized(batch, clientState, Permission.MODIFY, table1);
+
+        // MODIFY on table1 alone must not let the table2 statement through
+        grant(clientState, Permission.MODIFY, table1);
+        assertUnauthorized(batch, clientState, Permission.MODIFY, table2);
+
+        grant(clientState, Permission.MODIFY, table2);
+        batch.authorize(clientState);
+    }
+
+    /**
+     * The unconditional statement comes first and satisfies MODIFY for the table, but the conditional statement
+     * that follows must still require SELECT, since a CAS update can be used to simulate a read.
+     */
+    @Test
+    public void conditionalStatementAfterUnconditionalRequiresSelect()
+    {
+        ClientState clientState = createUserAndLogin();
+        BatchStatement batch = batch(clientState,
+                                     insert(table1, 0),
+                                     updateIf(table1, 0));
+
+        assertUnauthorized(batch, clientState, Permission.MODIFY, table1);
+
+        grant(clientState, Permission.MODIFY, table1);
+        assertUnauthorized(batch, clientState, Permission.SELECT, table1);
+
+        grant(clientState, Permission.SELECT, table1);
+        batch.authorize(clientState);
+    }
+
+    /**
+     * The view lookup is only performed for the first statement of each table, but the permissions it requires -
+     * SELECT on the base table, since the view update reads the current state, and MODIFY on every view - must
+     * still be enforced for the batch.
+     */
+    @Test
+    public void batchOnViewedTableRequiresModifyOnView()
+    {
+        // created last so that it is the current table the view is built on
+        String base = createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+        String view = createView("CREATE MATERIALIZED VIEW %s AS SELECT * FROM %s " +
+                                 "WHERE k IS NOT NULL AND v IS NOT NULL PRIMARY KEY (v, k)");
+
+        ClientState clientState = createUserAndLogin();
+        BatchStatement batch = batch(clientState,
+                                     insert(base, 0),
+                                     insert(base, 1));
+
+        assertUnauthorized(batch, clientState, Permission.MODIFY, base);
+
+        grant(clientState, Permission.MODIFY, base);
+        assertUnauthorized(batch, clientState, Permission.SELECT, base);
+
+        grant(clientState, Permission.SELECT, base);
+        assertUnauthorized(batch, clientState, Permission.MODIFY, view);
+
+        grant(clientState, Permission.MODIFY, view);
+        batch.authorize(clientState);
+    }
+
+    /**
+     * SELECT ensured for one base table must not carry over to the next one, so the state has to be cleared when
+     * the batch moves on to a different table.
+     */
+    @Test
+    public void batchOnSeveralViewedTablesRequiresSelectOnEachBase()
+    {
+        String base1 = createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+        String view1 = createView("CREATE MATERIALIZED VIEW %s AS SELECT * FROM %s " +
+                                  "WHERE k IS NOT NULL AND v IS NOT NULL PRIMARY KEY (v, k)");
+        String base2 = createTable("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+        String view2 = createView("CREATE MATERIALIZED VIEW %s AS SELECT * FROM %s " +
+                                  "WHERE k IS NOT NULL AND v IS NOT NULL PRIMARY KEY (v, k)");
+
+        ClientState clientState = createUserAndLogin();
+        BatchStatement batch = batch(clientState,
+                                     insert(base1, 0),
+                                     insert(base2, 0));
+
+        grant(clientState, Permission.MODIFY, base1);
+        grant(clientState, Permission.MODIFY, base2);
+        grant(clientState, Permission.SELECT, base1);
+        grant(clientState, Permission.MODIFY, view1);
+
+        // everything base1 needs is granted, but updating base2's view still requires reading base2
+        assertUnauthorized(batch, clientState, Permission.SELECT, base2);
+
+        grant(clientState, Permission.SELECT, base2);
+        assertUnauthorized(batch, clientState, Permission.MODIFY, view2);
+
+        grant(clientState, Permission.MODIFY, view2);
+        batch.authorize(clientState);
+    }
+
+    private BatchStatement batch(ClientState clientState, String... queries)
+    {
+        // prepared one by one, as they are when a batch arrives over the native protocol
+        List<ModificationStatement> statements = new ArrayList<>(queries.length);
+        for (String query : queries)
+            statements.add((ModificationStatement) QueryProcessor.getStatement(query, clientState));
+
+        return new BatchStatement(BatchStatement.Type.LOGGED, VariableSpecifications.empty(), statements, Attributes.none());
+    }
+
+    private static String insert(String table, int k)
+    {
+        return String.format("INSERT INTO %s.%s (k, v) VALUES (%d, 0)", KEYSPACE, table, k);
+    }
+
+    private static String updateIf(String table, int k)
+    {
+        return String.format("UPDATE %s.%s SET v = 1 WHERE k = %d IF v = 0", KEYSPACE, table, k);
+    }
+
+    private void grant(ClientState clientState, Permission permission, String table)
+    {
+        AuthTestUtils.authorize("GRANT %s ON TABLE %s.%s TO %s",
+                                permission, KEYSPACE, table, clientState.getUser().getName());
+    }
+
+    private void assertUnauthorized(BatchStatement batch, ClientState clientState, Permission permission, String table)
+    {
+        Assertions.assertThatThrownBy(() -> batch.authorize(clientState))
+                  .isInstanceOf(UnauthorizedException.class)
+                  .hasMessage("User %s has no %s permission on <table %s.%s> or any of its parents",
+                              clientState.getUser().getName(), permission, KEYSPACE, table);
+    }
+
+    private ClientState createUserAndLogin()
+    {
+        String username = AuthTestUtils.createName();
+        auth("CREATE ROLE %s WITH password = 'password' AND LOGIN = true", username);
+        ClientState clientState = ClientState.forExternalCalls(InetSocketAddress.createUnresolved("127.0.0.1", 123));
+        clientState.login(new AuthenticatedUser(username));
+        return clientState;
+    }
+}