Merge 'cql3: fix authorization bypass via BATCH prepared cache poisoning' from Marcin Maliszkiewicz

execute_batch_without_checking_exception_message() inserted entries
into the authorized prepared cache before verifying that
check_access() succeeded. A failed BATCH therefore left behind
cached 'authorized' entries that later let a direct EXECUTE of the
same prepared statement skip the authorization check entirely.

Move the cache insertion after the access check so that entries are
only cached on success. This matches the pattern already used by
do_execute_prepared() for individual EXECUTE requests.

Introduced in 98f5e49ea8

Fixes https://scylladb.atlassian.net/browse/SCYLLADB-1221

Backport: all supported versions

Closes scylladb/scylladb#29432

* github.com:scylladb/scylladb:
  test/cqlpy: add reproducer for BATCH prepared auth cache bypass
  cql3: fix authorization bypass via BATCH prepared cache poisoning

cql3/query_processor.cc

@@ -1071,6 +1071,11 @@ query_processor::execute_batch_without_checking_exception_message(
         query_options& options,
         std::unordered_map<prepared_cache_key_type, authorized_prepared_statements_cache::value_type> pending_authorization_entries) {
     auto access_future = co_await coroutine::as_future(batch->check_access(*this, query_state.get_client_state()));
+    bool failed = access_future.failed();
+    co_await audit::inspect(batch, query_state, options, failed);
+    if (failed) {
+        std::rethrow_exception(access_future.get_exception());
+    }
     co_await coroutine::parallel_for_each(pending_authorization_entries, [this, &query_state] (auto& e) -> future<> {
             try {
                 co_await _authorized_prepared_cache.insert(*query_state.get_client_state().user(), e.first, std::move(e.second));
@@ -1078,11 +1083,6 @@ query_processor::execute_batch_without_checking_exception_message(
                 log.error("failed to cache the entry: {}", std::current_exception());
             }
         });
-    bool failed = access_future.failed();
-    co_await audit::inspect(batch, query_state, options, failed);
-    if (access_future.failed()) {
-        std::rethrow_exception(access_future.get_exception());
-    }
     batch->validate();
     batch->validate(*this, query_state.get_client_state());
     _stats.queries_by_cl[size_t(options.get_consistency())] += batch->get_statements().size();

test/cqlpy/test_permissions.py

@@ -10,6 +10,7 @@ import pytest
 import time
 from . import rest_api
 from cassandra.protocol import SyntaxException, InvalidRequest, Unauthorized, ConfigurationException
+from cassandra.query import BatchStatement, BatchType
 from .util import new_test_table, new_function, new_user, new_session, new_test_keyspace, unique_name, new_type, new_materialized_view, is_scylla
 from contextlib import contextmanager
 
@@ -859,3 +860,35 @@ def test_select_system_table(cql):
             eventually_unauthorized(lambda: user1_session.execute(f'SELECT * FROM {roles_table}'))
             grant(cql, 'SELECT', roles_table, user1)
             eventually_authorized(lambda: user1_session.execute(f'SELECT * FROM {roles_table}'))
+
+# Reproducer for SCYLLADB-1221: a non-privileged user could bypass
+# authorization by exploiting the BATCH prepared cache population bug.
+# A failed BATCH execution was incorrectly populating the
+# authorized_prepared_cache, allowing subsequent direct executions of
+# the same prepared statement to skip the authorization check.
+def test_batch_prepared_auth_cache_bypass(cql, test_keyspace):
+    with new_test_table(cql, test_keyspace, "pk int PRIMARY KEY, val text") as table:
+        with new_user(cql) as username:
+            with new_session(cql, username) as user_session:
+                # The unprivileged user prepares an INSERT. Preparation itself
+                # succeeds — authorization is checked at execution time.
+                user_prepared = user_session.prepare(
+                    f"INSERT INTO {table} (pk, val) VALUES (?, ?)")
+
+                # Step 1: Verify direct execution is blocked (sanity check)
+                with pytest.raises(Unauthorized):
+                    user_session.execute(user_prepared, [42, "hacked"])
+
+                # Step 2: Execute via BATCH — this must fail with Unauthorized,
+                # but the bug causes the cache to be poisoned
+                batch = BatchStatement(batch_type=BatchType.LOGGED)
+                batch.add(user_prepared, [42, "hacked"])
+                with pytest.raises(Unauthorized):
+                    user_session.execute(batch)
+
+                # Step 3: Now execute the same prepared statement directly
+                # again. With the bug, the authorized_prepared_cache was
+                # populated during the failed BATCH, so no auth check
+                # happens and this succeeds.
+                with pytest.raises(Unauthorized):
+                    user_session.execute(user_prepared, [42, "hacked"])