diff --git a/cql3/query_processor.cc b/cql3/query_processor.cc
index 07afe72ee2..1cd6cb77de 100644
--- a/cql3/query_processor.cc
+++ b/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();
diff --git a/test/cqlpy/test_permissions.py b/test/cqlpy/test_permissions.py
index 2fc0753c8e..e4ec53f429 100644
--- a/test/cqlpy/test_permissions.py
+++ b/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"])