[Failure Store] Test API keys and skip_unavailable with RCS1 (#125782)

Adjust existing RCS1 tests to randomize using API keys for authorization
and `skip_unavailable` setting.

Followup on #125252

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/AbstractRemoteClusterSecurityFailureStoreRestIT.java

@@ -137,14 +137,14 @@ abstract class AbstractRemoteClusterSecurityFailureStoreRestIT extends AbstractR
         }
     }
 
-    protected Response performRequestWithRemoteSearchUser(final Request request) throws IOException {
+    protected static Response performRequestWithRemoteSearchUser(final Request request) throws IOException {
         request.setOptions(
             RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", headerFromRandomAuthMethod(REMOTE_SEARCH_USER, PASS))
         );
         return client().performRequest(request);
     }
 
-    protected Response performRequestWithUser(final String user, final Request request) throws IOException {
+    protected static Response performRequestWithUser(final String user, final Request request) throws IOException {
         request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", headerFromRandomAuthMethod(user, PASS)));
         return client().performRequest(request);
     }

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRCS1FailureStoreRestIT.java

@@ -8,24 +8,34 @@
 package org.elasticsearch.xpack.remotecluster;
 
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.settings.SecureString;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.cluster.FeatureFlag;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.elasticsearch.test.cluster.util.resource.Resource;
+import org.elasticsearch.test.rest.ObjectPath;
+import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.rules.RuleChain;
 import org.junit.rules.TestRule;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Locale;
+import java.util.Map;
+import java.util.function.Consumer;
 
+import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
 
 public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteClusterSecurityFailureStoreRestIT {
 
@@ -59,10 +69,190 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
     private static final String BACKING_FAILURE_STORE_INDEX_ACCESS = "backing_failure_store_index_access";
     private static final String BACKING_DATA_INDEX_ACCESS = "backing_data_index_access";
 
+    /**
+     * Maps usernames to their role descriptors. The usernames are also used as role names.
+     */
+    private static final Map<String, String> usersAndRolesOnFulfillingCluster = Map.of(DATA_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": ["test*"],
+              "privileges": ["read", "read_cross_cluster"]
+            }
+          ]
+        }""", FAILURE_STORE_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": ["test*", "non-existing-index"],
+              "privileges": ["read", "read_cross_cluster", "read_failure_store"]
+            }
+          ]
+        }""", MANAGE_FAILURE_STORE_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": ["test*", "non-existing-index"],
+              "privileges": ["manage_failure_store", "read_cross_cluster", "read_failure_store"]
+            }
+          ]
+        }""", ALL_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": ["*"],
+              "privileges": ["all"]
+            }
+          ]
+        }""", ONLY_READ_FAILURE_STORE_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": ["test*", "non-existing-index"],
+              "privileges": ["read_failure_store"]
+            }
+          ]
+        }""", BACKING_DATA_INDEX_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": [".ds-test*"],
+              "privileges": ["read", "read_cross_cluster"]
+            }
+          ]
+        }""", BACKING_FAILURE_STORE_INDEX_ACCESS, """
+        {
+          "indices": [
+            {
+              "names": [".fs-test*"],
+              "privileges": ["read", "read_cross_cluster"]
+            }
+          ]
+        }""");
+
+    /**
+     * The role must simply exist on query cluster, the actual access is irrelevant for user authorization.
+     * But we here grant access to the local and (for some also) remote indices to test mixed search
+     * and API key authorization.
+     */
+    private static final Map<String, String> usersAndRolesOnQueryCluster = Map.of(DATA_ACCESS, """
+        {
+          "cluster": ["manage_security"],
+          "indices": [
+             {
+              "names": ["local_index"],
+              "privileges": ["read"]
+            },
+            {
+              "names": ["test*"],
+              "privileges": ["read", "read_cross_cluster"]
+            }
+          ]
+        }""", FAILURE_STORE_ACCESS, """
+        {
+          "cluster": ["manage_security"],
+          "indices": [
+             {
+              "names": ["local_index"],
+              "privileges": ["read"]
+            },
+            {
+              "names": ["test*", "non-existing-index"],
+              "privileges": ["read", "read_cross_cluster", "read_failure_store"]
+            }
+          ]
+        }""", MANAGE_FAILURE_STORE_ACCESS, """
+        {
+          "cluster": ["manage_security"],
+          "indices": [
+            {
+              "names": ["local_index"],
+              "privileges": ["read"]
+            },
+            {
+              "names": ["test*", "non-existing-index"],
+              "privileges": ["manage_failure_store", "read_cross_cluster", "read_failure_store"]
+            }
+          ]
+        }""", ALL_ACCESS, """
+        {
+          "cluster": ["all"],
+          "indices": [
+            {
+              "names": ["*"],
+              "privileges": ["all"]
+            }
+          ]
+        }""", ONLY_READ_FAILURE_STORE_ACCESS, """
+        {
+          "cluster": ["manage_security"],
+          "indices": [
+            {
+              "names": ["test*", "non-existing-index"],
+              "privileges": ["read_failure_store"]
+            }
+          ]
+        }""", BACKING_DATA_INDEX_ACCESS, """
+        {
+          "cluster": ["manage_security"],
+          "indices": [
+            {
+              "names": [".ds-test*"],
+              "privileges": ["read", "read_cross_cluster"]
+            }
+          ]
+        }""", BACKING_FAILURE_STORE_INDEX_ACCESS, """
+        {
+          "cluster": ["manage_security"],
+          "indices": [
+            {
+              "names": [".fs-test*"],
+              "privileges": ["read", "read_cross_cluster"]
+            }
+          ]
+        }""");
+
+    /**
+     * Maps usernames to their API keys.
+     */
+    private static Map<String, Tuple<String, String>> apiKeys = new HashMap<>();
+
+    @Before
+    public void resetApiKeys() {
+        apiKeys = new HashMap<>();
+    }
+
+    private static void setupRoleAndUserOnFulfillingCluster() throws IOException {
+        for (var userAndRole : usersAndRolesOnFulfillingCluster.entrySet()) {
+            String roleAndUsername = userAndRole.getKey();
+            String roleDescriptor = userAndRole.getValue();
+            createRoleAndUserOnFulfillingCluster(roleAndUsername, roleDescriptor);
+        }
+    }
+
+    private static void setupUserAndRoleOnQueryCluster() throws IOException {
+        for (var userAndRole : usersAndRolesOnQueryCluster.entrySet()) {
+            String roleAndUsername = userAndRole.getKey();
+            String roleDescriptor = userAndRole.getValue();
+            createRoleOnQueryCluster(adminClient(), roleAndUsername, roleDescriptor);
+            createUserOnQueryCluster(adminClient(), roleAndUsername, PASS, roleAndUsername);
+            createAndStoreApiKeyOnQueryCluster(roleAndUsername, roleDescriptor);
+        }
+    }
+
+    /**
+     * Index some documents on the query cluster to use them in a mixed-cluster search.
+     */
+    private static void setupLocalDataOnQueryCluster() throws IOException {
+        final var indexDocRequest = new Request("POST", "/local_index/_doc?refresh=true");
+        indexDocRequest.setJsonEntity("{\"local_foo\": \"local_bar\"}");
+        assertOK(client().performRequest(indexDocRequest));
+    }
+
     public void testRCS1CrossClusterSearch() throws Exception {
         final boolean rcs1Security = true;
         final boolean isProxyMode = randomBoolean();
-        final boolean skipUnavailable = false; // we want to get actual failures and not skip and get empty results
+        final boolean skipUnavailable = randomBoolean();
         final boolean ccsMinimizeRoundtrips = randomBoolean();
 
         configureRemoteCluster(REMOTE_CLUSTER_ALIAS, fulfillingCluster, rcs1Security, isProxyMode, skipUnavailable);
@@ -86,11 +276,11 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
         testCcsWithDataSelectorNotSupported(ccsMinimizeRoundtrips);
         testCcsWithFailuresSelectorNotSupported(ccsMinimizeRoundtrips);
         testCcsWithoutSelectorsSupported(backingDataIndexName, ccsMinimizeRoundtrips);
-        testSearchingUnauthorizedIndices(otherBackingFailureIndexName, otherBackingDataIndexName, ccsMinimizeRoundtrips);
+        testSearchingUnauthorizedIndices(otherBackingFailureIndexName, otherBackingDataIndexName, ccsMinimizeRoundtrips, skipUnavailable);
         testSearchingWithAccessToAllIndices(ccsMinimizeRoundtrips, backingDataIndexName, otherBackingDataIndexName);
-        testBackingFailureIndexAccess(ccsMinimizeRoundtrips, backingFailureIndexName);
+        testBackingFailureIndexAccess(ccsMinimizeRoundtrips, backingFailureIndexName, skipUnavailable);
         testBackingDataIndexAccess(ccsMinimizeRoundtrips, backingDataIndexName);
-        testSearchingNonExistingIndices(ccsMinimizeRoundtrips);
+        testSearchingNonExistingIndices(ccsMinimizeRoundtrips, skipUnavailable);
         testResolveRemoteClustersIsUnauthorized();
     }
 
@@ -105,7 +295,7 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
             )
         );
         assertSearchResponseContainsIndices(
-            performRequestWithUser(BACKING_DATA_INDEX_ACCESS, dataIndexSearchRequest),
+            performRequestMaybeUsingApiKey(BACKING_DATA_INDEX_ACCESS, dataIndexSearchRequest),
             backingDataIndexName
         );
     }
@@ -130,35 +320,37 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
         final String[] expectedIndices = alsoSearchLocally
             ? new String[] { "local_index", backingDataIndexName, otherBackingDataIndexName }
             : new String[] { backingDataIndexName, otherBackingDataIndexName };
-        assertSearchResponseContainsIndices(performRequestWithUser(ALL_ACCESS, dataSearchRequest), expectedIndices);
+        assertSearchResponseContainsIndices(performRequestMaybeUsingApiKey(ALL_ACCESS, dataSearchRequest), expectedIndices);
     }
 
-    private void testSearchingNonExistingIndices(boolean ccsMinimizeRoundtrips) {
+    private void testSearchingNonExistingIndices(boolean ccsMinimizeRoundtrips, boolean skipUnavailable) {
         // searching non-existing index without permissions should result in 403
         {
-            final ResponseException exception = expectThrows(
-                ResponseException.class,
-                () -> performRequestWithUser(
+            final String indexToSearch = "non-existing-no-privileges";
+            final String action = ccsMinimizeRoundtrips ? "indices:data/read/search" : "indices:admin/search/search_shards";
+            executeAndAssert(
+                () -> performRequestMaybeUsingApiKey(
                     FAILURE_STORE_ACCESS,
                     new Request(
                         "GET",
                         String.format(
                             Locale.ROOT,
                             "/my_remote_cluster:%s/_search?ccs_minimize_roundtrips=%s",
-                            "non-existing-no-privileges",
+                            indexToSearch,
                             ccsMinimizeRoundtrips
                         )
                     )
-                )
+                ),
+                exception -> assertActionUnauthorized(exception, FAILURE_STORE_ACCESS, action, indexToSearch),
+                response -> assertActionUnauthorized(response, FAILURE_STORE_ACCESS, action, indexToSearch),
+                skipUnavailable
             );
-            final String action = ccsMinimizeRoundtrips ? "indices:data/read/search" : "indices:admin/search/search_shards";
-            assertActionUnauthorized(exception, FAILURE_STORE_ACCESS, action, "non-existing-no-privileges");
         }
         // searching non-existing index with permissions should result in 404
         {
-            final ResponseException exception = expectThrows(
-                ResponseException.class,
-                () -> performRequestWithUser(
+            final String indexToSearch = "non-existing-index";
+            executeAndAssert(
+                () -> performRequestMaybeUsingApiKey(
                     FAILURE_STORE_ACCESS,
                     new Request(
                         "GET",
@@ -169,22 +361,30 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
                             ccsMinimizeRoundtrips
                         )
                     )
-                )
+                ),
+                exception -> assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(404)),
+                response -> assertThat(
+                    ObjectPath.createFromResponse(response)
+                        .evaluate("_clusters.details.my_remote_cluster.failures.0.reason.reason")
+                        .toString(),
+                    containsString("no such index [" + indexToSearch + "]")
+                ),
+                skipUnavailable
             );
-            assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(404));
         }
     }
 
     private void testSearchingUnauthorizedIndices(
         String otherBackingFailureIndexName,
         String otherBackingDataIndexName,
-        boolean ccsMinimizeRoundtrips
+        boolean ccsMinimizeRoundtrips,
+        boolean skipUnavailable
     ) {
         // try searching remote index for which user has no access
         final String indexToSearch = randomFrom("other1", otherBackingFailureIndexName, otherBackingDataIndexName);
-        final ResponseException exception = expectThrows(
-            ResponseException.class,
-            () -> performRequestWithUser(
+        final String action = ccsMinimizeRoundtrips ? "indices:data/read/search" : "indices:admin/search/search_shards";
+        executeAndAssert(
+            () -> performRequestMaybeUsingApiKey(
                 FAILURE_STORE_ACCESS,
                 new Request(
                     "GET",
@@ -195,13 +395,15 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
                         ccsMinimizeRoundtrips
                     )
                 )
-            )
+            ),
+            exception -> assertActionUnauthorized(exception, FAILURE_STORE_ACCESS, action, indexToSearch),
+            response -> assertActionUnauthorized(response, FAILURE_STORE_ACCESS, action, indexToSearch),
+            skipUnavailable
         );
-        final String action = ccsMinimizeRoundtrips ? "indices:data/read/search" : "indices:admin/search/search_shards";
-        assertActionUnauthorized(exception, FAILURE_STORE_ACCESS, action, indexToSearch);
     }
 
-    private void testBackingFailureIndexAccess(boolean ccsMinimizeRoundtrips, String backingFailureIndexName) throws IOException {
+    private void testBackingFailureIndexAccess(boolean ccsMinimizeRoundtrips, String backingFailureIndexName, boolean skipUnavailable)
+        throws IOException {
         // direct access to backing failure index is subject to the user's permissions
         // it might fail in some cases and work in others
         Request failureIndexSearchRequest = new Request(
@@ -215,16 +417,17 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
         );
 
         // user with access to all should be able to search the backing failure index
-        assertSearchResponseContainsIndices(performRequestWithUser(ALL_ACCESS, failureIndexSearchRequest), backingFailureIndexName);
+        assertSearchResponseContainsIndices(performRequestMaybeUsingApiKey(ALL_ACCESS, failureIndexSearchRequest), backingFailureIndexName);
 
         // user with data only access should not be able to search the backing failure index
         {
-            final ResponseException exception = expectThrows(
-                ResponseException.class,
-                () -> performRequestWithUser(DATA_ACCESS, failureIndexSearchRequest)
-            );
             final String action = ccsMinimizeRoundtrips ? "indices:data/read/search" : "indices:admin/search/search_shards";
-            assertActionUnauthorized(exception, DATA_ACCESS, action, backingFailureIndexName);
+            executeAndAssert(
+                () -> performRequestMaybeUsingApiKey(DATA_ACCESS, failureIndexSearchRequest),
+                exception -> assertActionUnauthorized(exception, DATA_ACCESS, action, backingFailureIndexName),
+                response -> assertActionUnauthorized(response, DATA_ACCESS, action, backingFailureIndexName),
+                skipUnavailable
+            );
         }
 
         // for user with access to failure store, it depends on the underlying action that is being sent to the remote cluster
@@ -235,7 +438,7 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
             // from a security perspective, this is a valid use case and there is no way to prevent this with RCS1 security model
             // since from the fulfilling cluster perspective this request is no different from any other local search request
             assertSearchResponseContainsIndices(
-                performRequestWithUser(FAILURE_STORE_ACCESS, failureIndexSearchRequest),
+                performRequestMaybeUsingApiKey(FAILURE_STORE_ACCESS, failureIndexSearchRequest),
                 backingFailureIndexName
             );
         } else {
@@ -244,21 +447,32 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
             // which does not grant access to the indices:admin/search/search_shards action
             // this action is granted by read_cross_cluster privilege which is currently
             // not supporting the failure backing indices (only data backing indices)
-            final ResponseException exception = expectThrows(
-                ResponseException.class,
-                () -> performRequestWithUser(FAILURE_STORE_ACCESS, failureIndexSearchRequest)
+            executeAndAssert(
+                () -> performRequestMaybeUsingApiKey(FAILURE_STORE_ACCESS, failureIndexSearchRequest),
+                exception -> assertActionUnauthorized(
+                    exception,
+                    FAILURE_STORE_ACCESS,
+                    "indices:admin/search/search_shards",
+                    backingFailureIndexName
+                ),
+                response -> assertActionUnauthorized(
+                    response,
+                    FAILURE_STORE_ACCESS,
+                    "indices:admin/search/search_shards",
+                    backingFailureIndexName
+                ),
+                skipUnavailable
             );
-            assertActionUnauthorized(exception, FAILURE_STORE_ACCESS, "indices:admin/search/search_shards", backingFailureIndexName);
         }
 
         // user with manage failure store access should be able to search the backing failure index
         assertSearchResponseContainsIndices(
-            performRequestWithUser(MANAGE_FAILURE_STORE_ACCESS, failureIndexSearchRequest),
+            performRequestMaybeUsingApiKey(MANAGE_FAILURE_STORE_ACCESS, failureIndexSearchRequest),
             backingFailureIndexName
         );
 
         assertSearchResponseContainsIndices(
-            performRequestWithUser(BACKING_FAILURE_STORE_INDEX_ACCESS, failureIndexSearchRequest),
+            performRequestMaybeUsingApiKey(BACKING_FAILURE_STORE_INDEX_ACCESS, failureIndexSearchRequest),
             backingFailureIndexName
         );
 
@@ -282,7 +496,7 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
             final String[] expectedIndices = alsoSearchLocally
                 ? new String[] { "local_index", backingDataIndexName }
                 : new String[] { backingDataIndexName };
-            assertSearchResponseContainsIndices(performRequestWithUser(user, dataSearchRequest), expectedIndices);
+            assertSearchResponseContainsIndices(performRequestMaybeUsingApiKey(user, dataSearchRequest), expectedIndices);
         }
     }
 
@@ -303,7 +517,7 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
             );
             final ResponseException exception = expectThrows(
                 ResponseException.class,
-                () -> performRequestWithUser(user, dataSearchRequest)
+                () -> performRequestMaybeUsingApiKey(user, dataSearchRequest)
             );
             assertSelectorsNotSupported(exception);
         }
@@ -322,7 +536,7 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
             // query remote cluster using ::failures selector should fail (regardless of the user's permissions)
             final ResponseException exception = expectThrows(
                 ResponseException.class,
-                () -> performRequestWithUser(
+                () -> performRequestMaybeUsingApiKey(
                     user,
                     new Request(
                         "GET",
@@ -343,97 +557,40 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
         // user with only read_failure_store access should not be able to resolve remote clusters
         var exc = expectThrows(
             ResponseException.class,
-            () -> performRequestWithUser(ONLY_READ_FAILURE_STORE_ACCESS, new Request("GET", "/_resolve/cluster/" + REMOTE_CLUSTER_ALIAS))
+            () -> performRequestMaybeUsingApiKey(ONLY_READ_FAILURE_STORE_ACCESS, new Request("GET", "/_resolve/cluster"))
         );
         assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(403));
         assertThat(
             exc.getMessage(),
-            containsString(
-                "action ["
-                    + "indices:admin/resolve/cluster"
-                    + "] is unauthorized for user ["
-                    + ONLY_READ_FAILURE_STORE_ACCESS
-                    + "] "
-                    + "with effective roles ["
-                    + ONLY_READ_FAILURE_STORE_ACCESS
-                    + "]"
+            anyOf(
+                containsString(
+                    "action ["
+                        + "indices:admin/resolve/cluster"
+                        + "] is unauthorized for user ["
+                        + ONLY_READ_FAILURE_STORE_ACCESS
+                        + "] "
+                        + "with effective roles ["
+                        + ONLY_READ_FAILURE_STORE_ACCESS
+                        + "]"
+                ),
+                containsString(
+                    "action [indices:admin/resolve/cluster] is unauthorized for API key id ["
+                        + apiKeys.get(ONLY_READ_FAILURE_STORE_ACCESS).v1()
+                        + "] of user ["
+                        + ONLY_READ_FAILURE_STORE_ACCESS
+                        + "]"
+                )
             )
         );
     }
 
-    private static void setupLocalDataOnQueryCluster() throws IOException {
-        // Index some documents, to use them in a mixed-cluster search
-        final var indexDocRequest = new Request("POST", "/local_index/_doc?refresh=true");
-        indexDocRequest.setJsonEntity("{\"local_foo\": \"local_bar\"}");
-        assertOK(client().performRequest(indexDocRequest));
-    }
-
-    private static void setupUserAndRoleOnQueryCluster() throws IOException {
-        createRole(adminClient(), ALL_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["*"],
-                  "privileges": ["all"]
-                }
-              ]
-            }""");
-        createUser(adminClient(), ALL_ACCESS, PASS, ALL_ACCESS);
-        // the role must simply exist on query cluster, the access is irrelevant,
-        // but we here grant the access to local_index only to test mixed search
-        createRole(adminClient(), FAILURE_STORE_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["local_index"],
-                  "privileges": ["read"]
-                }
-              ]
-            }""");
-        createUser(adminClient(), FAILURE_STORE_ACCESS, PASS, FAILURE_STORE_ACCESS);
-        createRole(adminClient(), DATA_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["local_index"],
-                  "privileges": ["read"]
-                }
-              ]
-            }""");
-        createUser(adminClient(), DATA_ACCESS, PASS, DATA_ACCESS);
-        createRole(adminClient(), MANAGE_FAILURE_STORE_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["local_index"],
-                  "privileges": ["read"]
-                }
-              ]
-            }""");
-        createUser(adminClient(), MANAGE_FAILURE_STORE_ACCESS, PASS, MANAGE_FAILURE_STORE_ACCESS);
-
-        createRole(adminClient(), ONLY_READ_FAILURE_STORE_ACCESS, """
-            {
-            }""");
-        createUser(adminClient(), ONLY_READ_FAILURE_STORE_ACCESS, PASS, ONLY_READ_FAILURE_STORE_ACCESS);
-        createRole(adminClient(), BACKING_FAILURE_STORE_INDEX_ACCESS, """
-            {
-            }""");
-        createUser(adminClient(), BACKING_FAILURE_STORE_INDEX_ACCESS, PASS, BACKING_FAILURE_STORE_INDEX_ACCESS);
-        createRole(adminClient(), BACKING_DATA_INDEX_ACCESS, """
-            {
-            }""");
-        createUser(adminClient(), BACKING_DATA_INDEX_ACCESS, PASS, BACKING_DATA_INDEX_ACCESS);
-
-    }
-
-    private static void createRole(RestClient client, String role, String roleDescriptor) throws IOException {
+    private static void createRoleOnQueryCluster(RestClient client, String role, String roleDescriptor) throws IOException {
         final Request putRoleRequest = new Request("PUT", "/_security/role/" + role);
         putRoleRequest.setJsonEntity(roleDescriptor);
         assertOK(client.performRequest(putRoleRequest));
     }
 
-    private static void createUser(RestClient client, String user, SecureString password, String role) throws IOException {
+    private static void createUserOnQueryCluster(RestClient client, String user, SecureString password, String role) throws IOException {
         final Request putUserRequest = new Request("PUT", "/_security/user/" + user);
         putUserRequest.setJsonEntity(Strings.format("""
             {
@@ -443,83 +600,9 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
         assertOK(client.performRequest(putUserRequest));
     }
 
-    private static void setupRoleAndUserOnFulfillingCluster() throws IOException {
-        putRoleOnFulfillingCluster(DATA_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["test*"],
-                  "privileges": ["read", "read_cross_cluster"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(DATA_ACCESS, DATA_ACCESS);
-
-        putRoleOnFulfillingCluster(FAILURE_STORE_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["test*", "non-existing-index"],
-                  "privileges": ["read", "read_cross_cluster", "read_failure_store"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(FAILURE_STORE_ACCESS, FAILURE_STORE_ACCESS);
-
-        putRoleOnFulfillingCluster(MANAGE_FAILURE_STORE_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["test*", "non-existing-index"],
-                  "privileges": ["manage_failure_store", "read_cross_cluster", "read_failure_store"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(MANAGE_FAILURE_STORE_ACCESS, MANAGE_FAILURE_STORE_ACCESS);
-
-        putRoleOnFulfillingCluster(ALL_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["*"],
-                  "privileges": ["all"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(ALL_ACCESS, ALL_ACCESS);
-
-        putRoleOnFulfillingCluster(ONLY_READ_FAILURE_STORE_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": ["test*", "non-existing-index"],
-                  "privileges": ["read_failure_store"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(ONLY_READ_FAILURE_STORE_ACCESS, ONLY_READ_FAILURE_STORE_ACCESS);
-
-        putRoleOnFulfillingCluster(BACKING_DATA_INDEX_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": [".ds-test*"],
-                  "privileges": ["read", "read_cross_cluster"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(BACKING_DATA_INDEX_ACCESS, BACKING_DATA_INDEX_ACCESS);
-
-        putRoleOnFulfillingCluster(BACKING_FAILURE_STORE_INDEX_ACCESS, """
-            {
-              "indices": [
-                {
-                  "names": [".fs-test*"],
-                  "privileges": ["read", "read_cross_cluster"]
-                }
-              ]
-            }""");
-        putUserOnFulfillingCluster(BACKING_FAILURE_STORE_INDEX_ACCESS, BACKING_FAILURE_STORE_INDEX_ACCESS);
+    private static void createRoleAndUserOnFulfillingCluster(String userAndRoleName, String roleDescriptor) throws IOException {
+        putRoleOnFulfillingCluster(userAndRoleName, roleDescriptor);
+        putUserOnFulfillingCluster(userAndRoleName, userAndRoleName);
     }
 
     private static void putRoleOnFulfillingCluster(String roleName, String roleDescriptor) throws IOException {
@@ -547,18 +630,139 @@ public class RemoteClusterSecurityRCS1FailureStoreRestIT extends AbstractRemoteC
         assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(403));
         assertThat(
             exception.getMessage(),
-            containsString(
-                "action ["
-                    + action
-                    + "] is unauthorized for user ["
-                    + userAndRole
-                    + "] "
-                    + "with effective roles ["
-                    + userAndRole
-                    + "] on indices ["
-                    + backingFailureIndexName
-                    + "]"
+            anyOf(
+                containsString(
+                    "action ["
+                        + action
+                        + "] is unauthorized for user ["
+                        + userAndRole
+                        + "] "
+                        + "with effective roles ["
+                        + userAndRole
+                        + "] on indices ["
+                        + backingFailureIndexName
+                        + "]"
+                ),
+                containsString(
+                    "action ["
+                        + action
+                        + "] is unauthorized for API key id ["
+                        + (apiKeys.containsKey(userAndRole)
+                            ? apiKeys.get(userAndRole).v1()
+                            : "<there is no test API key for this user - ignore this assertion>")
+                        + "] of user ["
+                        + userAndRole
+                        + "] on indices ["
+                        + backingFailureIndexName
+                        + "]"
+                )
             )
         );
     }
+
+    private static void assertActionUnauthorized(Response response, String userAndRole, String action, String backingFailureIndexName)
+        throws IOException {
+        assertThat(response.getStatusLine().getStatusCode(), equalTo(200));
+        assertThat(
+            ObjectPath.createFromResponse(response).evaluate("_clusters.details.my_remote_cluster.failures.0.reason.reason").toString(),
+            anyOf(
+                containsString(
+                    "action ["
+                        + action
+                        + "] is unauthorized for user ["
+                        + userAndRole
+                        + "] "
+                        + "with effective roles ["
+                        + userAndRole
+                        + "] on indices ["
+                        + backingFailureIndexName
+                        + "]"
+                ),
+                containsString(
+                    "action ["
+                        + action
+                        + "] is unauthorized for API key id ["
+                        + (apiKeys.containsKey(userAndRole)
+                            ? apiKeys.get(userAndRole).v1()
+                            : "<there is no test API key for this user - ignore this assertion>")
+                        + "] of user ["
+                        + userAndRole
+                        + "] on indices ["
+                        + backingFailureIndexName
+                        + "]"
+                )
+            )
+        );
+    }
+
+    protected static void createAndStoreApiKeyOnQueryCluster(String user, @Nullable String roleDescriptors) throws IOException {
+        assertThat("API key already registered for user: " + user, apiKeys.containsKey(user), is(false));
+        apiKeys.put(user, createApiKeyOnQueryCluster(user, roleDescriptors));
+    }
+
+    private static Tuple<String, String> createApiKeyOnQueryCluster(String user, String roleDescriptors) throws IOException {
+        var request = new Request("POST", "/_security/api_key");
+        if (roleDescriptors == null) {
+            request.setJsonEntity("""
+                {
+                    "name": "test-api-key"
+                }
+                """);
+        } else {
+            request.setJsonEntity(org.elasticsearch.common.Strings.format("""
+                {
+                    "name": "test-api-key",
+                    "role_descriptors": {
+                        "%s": %s
+                    }
+                }
+                """, user, roleDescriptors));
+        }
+        request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", basicAuthHeaderValue(user, PASS)).build());
+        Response response = client().performRequest(request);
+        assertOK(response);
+        Map<String, Object> responseAsMap = responseAsMap(response);
+        return new Tuple<>((String) responseAsMap.get("id"), (String) responseAsMap.get("encoded"));
+    }
+
+    protected static Response performRequestMaybeUsingApiKey(String user, Request request) throws IOException {
+        if (randomBoolean() && apiKeys.containsKey(user)) {
+            return performRequestWithApiKey(apiKeys.get(user).v2(), request);
+        } else {
+            return performRequestWithUser(user, request);
+        }
+    }
+
+    private static Response performRequestWithApiKey(String apiKey, Request request) throws IOException {
+        request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", "ApiKey " + apiKey).build());
+        return client().performRequest(request);
+    }
+
+    private static <T extends Throwable> void executeAndAssert(
+        ThrowableCommand<Response> requestCommand,
+        Consumer<ResponseException> exceptionAssertion,
+        ThrowableConsumer<Response> responseAssertion,
+        boolean expectResponse
+    ) {
+        if (expectResponse) {
+            try {
+                responseAssertion.accept(requestCommand.execute());
+            } catch (Exception e) {
+                fail(e, "Not expected exception to be thrown: " + e.getMessage());
+            }
+        } else {
+            exceptionAssertion.accept(expectThrows(ResponseException.class, requestCommand::execute));
+        }
+    }
+
+    @FunctionalInterface
+    private interface ThrowableCommand<T> {
+        T execute() throws Exception;
+    }
+
+    @FunctionalInterface
+    public interface ThrowableConsumer<T> {
+
+        void accept(T t) throws Exception;
+    }
 }