Cross-cluster painless/execute actions should check permissions only on target remote cluster (#105360)

Fixes issue #105084, where the cross-cluster painless execute API inappropriately throws a security exception (on a secured cluster) if permissions for the remote index target are not present in the local querying cluster. Since it is a remote request, the security check should be happening on the remote cluster, not the local one.

We were not able to simply implement IndicesRequest.Replaceable to solve this case, as the painless execute API does not support wildcards and implementing Replaceable takes the security checks and index resolutions down paths that don't make sense for this use case.

A new interface, IndicesRequest.SingleIndexNoWildcards was created to support the specific use case that allows remote indices, but no wildcards. Changes were made to RBACEngine and IndiciesAndAliasesResolver to handle this new interface appropriately.

Two new IT tests were added - one covering clusters secured with RCS1 and the other clusters secured with RCS2. Both cover the bug fixed in this commit.

docs/changelog/105360.yaml

@@ -0,0 +1,6 @@
+pr: 105360
+summary: Cross-cluster painless/execute actions should check permissions only on target
+  remote cluster
+area: Search
+type: bug
+issues: []

modules/lang-painless/src/main/java/org/elasticsearch/painless/action/PainlessExecuteAction.java

@@ -24,6 +24,7 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.RemoteClusterActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -91,6 +92,7 @@ import org.elasticsearch.script.StringFieldScript;
 import org.elasticsearch.search.lookup.SearchLookup;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
@@ -102,6 +104,7 @@ import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -121,7 +124,7 @@ public class PainlessExecuteAction {
 
     private PainlessExecuteAction() {/* no instances */}
 
-    public static class Request extends SingleShardRequest<Request> implements ToXContentObject {
+    public static class Request extends SingleShardRequest<Request> implements ToXContentObject, IndicesRequest.SingleIndexNoWildcards {
 
         private static final ParseField SCRIPT_FIELD = new ParseField("script");
         private static final ParseField CONTEXT_FIELD = new ParseField("context");
@@ -217,7 +220,9 @@ public class PainlessExecuteAction {
             }
 
             /**
-             * @param indexExpression should be of the form "index" or "cluster:index". Wildcards are OK.
+             * @param indexExpression should be of the form "index" or "cluster:index". Wildcards are not allowed
+             *                        (if wildcards are present, an exception will be thrown in later processing, so
+             *                        we don't check it here).
              * @return Tuple where first entry is clusterAlias, which will be null if not in the indexExpression
              *         and second entry is the index name
              *         Tuple(null, null) will be returned if indexExpression is null
@@ -229,7 +234,8 @@ public class PainlessExecuteAction {
                     return new Tuple<>(null, null);
                 }
                 String trimmed = indexExpression.trim();
-                if (trimmed.startsWith(":") || trimmed.endsWith(":")) {
+                String sep = String.valueOf(RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR);
+                if (trimmed.startsWith(sep) || trimmed.endsWith(sep)) {
                     throw new IllegalArgumentException(
                         "Unable to parse one single valid index name from the provided index: [" + indexExpression + "]"
                     );
@@ -241,10 +247,10 @@ public class PainlessExecuteAction {
                 // Instead, it will fail with the inaccurate and confusing error message:
                 // "Cross-cluster calls are not supported in this context but remote indices were requested: [blogs,remote1:blogs]"
                 // which comes later out of the IndexNameExpressionResolver pathway this code uses.
-                String[] parts = indexExpression.split(":", 2);
+                String[] parts = indexExpression.split(sep, 2);
                 if (parts.length == 1) {
                     return new Tuple<>(null, parts[0]);
-                } else if (parts.length == 2 && parts[1].contains(":") == false) {
+                } else if (parts.length == 2 && parts[1].contains(sep) == false) {
                     return new Tuple<>(parts[0], parts[1]);
                 } else {
                     throw new IllegalArgumentException(
@@ -358,7 +364,11 @@ public class PainlessExecuteAction {
             this.context = scriptContextName != null ? fromScriptContextName(scriptContextName) : PainlessTestScript.CONTEXT;
             if (setup != null) {
                 this.contextSetup = setup;
-                index(contextSetup.index);
+                if (contextSetup.getClusterAlias() == null) {
+                    index(contextSetup.getIndex());
+                } else {
+                    index(contextSetup.getClusterAlias() + RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR + contextSetup.getIndex());
+                }
             } else {
                 contextSetup = null;
             }
@@ -527,14 +537,34 @@ public class PainlessExecuteAction {
             if (request.getContextSetup() == null || request.getContextSetup().getClusterAlias() == null) {
                 super.doExecute(task, request, listener);
             } else {
-                // forward to remote cluster
-                String clusterAlias = request.getContextSetup().getClusterAlias();
+                // forward to remote cluster after stripping off the clusterAlias from the index expression
+                removeClusterAliasFromIndexExpression(request);
                 transportService.getRemoteClusterService()
-                    .getRemoteClusterClient(clusterAlias, EsExecutors.DIRECT_EXECUTOR_SERVICE)
+                    .getRemoteClusterClient(request.getContextSetup().getClusterAlias(), EsExecutors.DIRECT_EXECUTOR_SERVICE)
                     .execute(PainlessExecuteAction.REMOTE_TYPE, request, listener);
             }
         }
 
+        // Visible for testing
+        static void removeClusterAliasFromIndexExpression(Request request) {
+            if (request.index() != null) {
+                String[] split = request.index().split(String.valueOf(RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR));
+                if (split.length > 1) {
+                    /*
+                     * if the cluster alias is null and the index field has a clusterAlias (clusterAlias:index notation)
+                     * that means this is executing on a remote cluster (it was forwarded by the querying cluster).
+                     * The clusterAlias is not Writeable, so it will be null in the ContextSetup on the remote cluster.
+                     * We need to strip off the clusterAlias from the index before executing the script locally,
+                     * so it will resolve to a local index
+                     */
+                    assert split.length == 2
+                        : "If the index contains the REMOTE_CLUSTER_INDEX_SEPARATOR it should have only two parts but it has "
+                            + Arrays.toString(split);
+                    request.index(split[1]);
+                }
+            }
+        }
+
         @Override
         protected Writeable.Reader<Response> getResponseReader() {
             return Response::new;

modules/lang-painless/src/test/java/org/elasticsearch/painless/action/PainlessExecuteApiTests.java

@@ -36,6 +36,7 @@ import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static org.elasticsearch.painless.action.PainlessExecuteAction.TransportAction.innerShardOperation;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
 
 public class PainlessExecuteApiTests extends ESSingleNodeTestCase {
 
@@ -515,4 +516,43 @@ public class PainlessExecuteApiTests extends ESSingleNodeTestCase {
         expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("remote1:foo,remote2:bar"));
         expectThrows(IllegalArgumentException.class, () -> Request.ContextSetup.parseClusterAliasAndIndex("a:b,c:d,e:f"));
     }
+
+    public void testRemoveClusterAliasFromIndexExpression() {
+        {
+            // index expressions with no clusterAlias should come back unchanged
+            PainlessExecuteAction.Request request = createRequest("blogs");
+            assertThat(request.index(), equalTo("blogs"));
+            PainlessExecuteAction.TransportAction.removeClusterAliasFromIndexExpression(request);
+            assertThat(request.index(), equalTo("blogs"));
+        }
+        {
+            // index expressions with no index specified should come back unchanged
+            PainlessExecuteAction.Request request = createRequest(null);
+            assertThat(request.index(), nullValue());
+            PainlessExecuteAction.TransportAction.removeClusterAliasFromIndexExpression(request);
+            assertThat(request.index(), nullValue());
+        }
+        {
+            // index expressions with clusterAlias should come back with it stripped off
+            PainlessExecuteAction.Request request = createRequest("remote1:blogs");
+            assertThat(request.index(), equalTo("remote1:blogs"));
+            PainlessExecuteAction.TransportAction.removeClusterAliasFromIndexExpression(request);
+            assertThat(request.index(), equalTo("blogs"));
+        }
+        {
+            // index expressions with clusterAlias should come back with it stripped off
+            PainlessExecuteAction.Request request = createRequest("remote1:remote1");
+            assertThat(request.index(), equalTo("remote1:remote1"));
+            PainlessExecuteAction.TransportAction.removeClusterAliasFromIndexExpression(request);
+            assertThat(request.index(), equalTo("remote1"));
+        }
+    }
+
+    private PainlessExecuteAction.Request createRequest(String indexExpression) {
+        return new PainlessExecuteAction.Request(
+            new Script("100.0 / 1000.0"),
+            null,
+            new PainlessExecuteAction.Request.ContextSetup(indexExpression, null, null)
+        );
+    }
 }

server/src/main/java/org/elasticsearch/action/IndicesRequest.java

@@ -59,4 +59,17 @@ public interface IndicesRequest {
             return false;
         }
     }
+
+    /**
+     * For use cases where a Request instance cannot implement Replaceable due to not supporting wildcards
+     * and only supporting a single index at a time, this is an alternative interface that the
+     * security layer checks against to determine if remote indices are allowed for that Request type.
+     *
+     * This may change with https://github.com/elastic/elasticsearch/issues/105598
+     */
+    interface SingleIndexNoWildcards extends IndicesRequest {
+        default boolean allowsRemoteIndices() {
+            return true;
+        }
+    }
 }

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

@@ -0,0 +1,223 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.remotecluster;
+
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.junit.ClassRule;
+import org.junit.rules.RuleChain;
+import org.junit.rules.TestRule;
+
+import java.io.IOException;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ * Tests cross-cluster painless/execute API under RCS1.0 security model
+ */
+public class RemoteClusterSecurityRCS1PainlessExecuteIT extends AbstractRemoteClusterSecurityTestCase {
+
+    static {
+        fulfillingCluster = ElasticsearchCluster.local().name("fulfilling-cluster").nodes(3).apply(commonClusterConfig).build();
+
+        queryCluster = ElasticsearchCluster.local().name("query-cluster").apply(commonClusterConfig).build();
+    }
+
+    @ClassRule
+    public static TestRule clusterRule = RuleChain.outerRule(fulfillingCluster).around(queryCluster);
+
+    @SuppressWarnings({ "unchecked", "checkstyle:LineLength" })
+    public void testPainlessExecute() throws Exception {
+        // Setup RCS 1.0 (basicSecurity=true)
+        configureRemoteCluster("my_remote_cluster", fulfillingCluster, true, randomBoolean(), randomBoolean());
+        {
+            // Query cluster -> add role for test user - do not give any privileges for remote_indices
+            final var putRoleRequest = new Request("PUT", "/_security/role/" + REMOTE_SEARCH_ROLE);
+            putRoleRequest.setJsonEntity("""
+                {
+                  "indices": [
+                    {
+                      "names": ["local_index", "my_local*"],
+                      "privileges": ["read"]
+                    }
+                  ]
+                }""");
+            assertOK(adminClient().performRequest(putRoleRequest));
+
+            // Query cluster -> create user and assign role
+            final var putUserRequest = new Request("PUT", "/_security/user/" + REMOTE_SEARCH_USER);
+            putUserRequest.setJsonEntity("""
+                {
+                  "password": "x-pack-test-password",
+                  "roles" : ["remote_search"]
+                }""");
+            assertOK(adminClient().performRequest(putUserRequest));
+
+            // Query cluster -> create test index
+            final var indexDocRequest = new Request("POST", "/local_index/_doc?refresh=true");
+            indexDocRequest.setJsonEntity("{\"local_foo\": \"local_bar\"}");
+            assertOK(client().performRequest(indexDocRequest));
+
+            // Fulfilling cluster -> create test indices
+            final Request bulkRequest = new Request("POST", "/_bulk?refresh=true");
+            bulkRequest.setJsonEntity(Strings.format("""
+                { "index": { "_index": "index1" } }
+                { "foo": "bar" }
+                { "index": { "_index": "secretindex" } }
+                { "bar": "foo" }
+                """));
+            assertOK(performRequestAgainstFulfillingCluster(bulkRequest));
+        }
+
+        {
+            // TEST CASE 1: Query local cluster for local_index - should work since role has read perms for it
+            Request painlessExecuteLocal = createPainlessExecuteRequest("local_index");
+            Response response = performRequestWithRemoteSearchUser(painlessExecuteLocal);
+            assertOK(response);
+            String responseBody = EntityUtils.toString(response.getEntity());
+            assertThat(responseBody, equalTo("{\"result\":[\"test\"]}"));
+        }
+        {
+            // TEST CASE 2: Query remote cluster for index1 - should fail since no permissions granted for remote clusters yet
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index1");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [index1]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // add user role and user on remote cluster
+            var putRoleOnRemoteClusterRequest = new Request("PUT", "/_security/role/" + REMOTE_SEARCH_ROLE);
+            putRoleOnRemoteClusterRequest.setJsonEntity("""
+                {
+                  "indices": [
+                    {
+                      "names": ["index*"],
+                      "privileges": ["read", "read_cross_cluster"]
+                    }
+                  ]
+                }""");
+            assertOK(performRequestAgainstFulfillingCluster(putRoleOnRemoteClusterRequest));
+
+            var putUserOnRemoteClusterRequest = new Request("PUT", "/_security/user/" + REMOTE_SEARCH_USER);
+            putUserOnRemoteClusterRequest.setJsonEntity("""
+                {
+                  "password": "x-pack-test-password",
+                  "roles" : ["remote_search"]
+                }""");
+            assertOK(performRequestAgainstFulfillingCluster(putUserOnRemoteClusterRequest));
+        }
+        {
+            // TEST CASE 3: Query remote cluster for secretindex - should fail since no perms granted for it
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:secretindex");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [secretindex]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // TEST CASE 4: Query remote cluster for index1 - should succeed since read and cross-cluster-read perms granted
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index1");
+            Response response = performRequestWithRemoteSearchUser(painlessExecuteRemote);
+            String responseBody = EntityUtils.toString(response.getEntity());
+            assertOK(response);
+            assertThat(responseBody, equalTo("{\"result\":[\"test\"]}"));
+        }
+        {
+            // TEST CASE 5: Query local cluster for not_present index - should fail with 403 since role does not have perms for this index
+            Request painlessExecuteLocal = createPainlessExecuteRequest("index_not_present");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteLocal));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [index_not_present]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // TEST CASE 6: Query local cluster for my_local_123 index - role has perms for this pattern, but index does not exist, so 404
+            Request painlessExecuteLocal = createPainlessExecuteRequest("my_local_123");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteLocal));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(404));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("\"type\":\"index_not_found_exception\""));
+        }
+        {
+            // TEST CASE 7: Query local cluster for my_local* index - painless/execute does not allow wildcards, so fails with 400
+            Request painlessExecuteLocal = createPainlessExecuteRequest("my_local*");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteLocal));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(400));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("indices:data/read/scripts/painless/execute does not support wildcards"));
+            assertThat(errorResponseBody, containsString("\"type\":\"illegal_argument_exception\""));
+        }
+        {
+            // TEST CASE 8: Query remote cluster for cluster that does not exist, and user does not have perms for that pattern - 403 ???
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:abc123");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [abc123]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // TEST CASE 9: Query remote cluster for cluster that does not exist, but has permissions for the index pattern - 404
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index123");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(404));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("\"type\":\"index_not_found_exception\""));
+        }
+        {
+            // TEST CASE 10: Query remote cluster with wildcard in index - painless/execute does not allow wildcards, so fails with 400
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index*");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(400));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("indices:data/read/scripts/painless/execute does not support wildcards"));
+            assertThat(errorResponseBody, containsString("\"type\":\"illegal_argument_exception\""));
+        }
+    }
+
+    private static Request createPainlessExecuteRequest(String indexExpression) {
+        Request painlessExecuteLocal = new Request("POST", "_scripts/painless/_execute");
+        String body = """
+            {
+                "script": {
+                    "source": "emit(\\"test\\")"
+                },
+                "context": "keyword_field",
+                "context_setup": {
+                    "index": "INDEX_EXPRESSION_HERE",
+                    "document": {
+                        "@timestamp": "2023-05-06T16:22:22.000Z"
+                    }
+                }
+            }""".replace("INDEX_EXPRESSION_HERE", indexExpression);
+        painlessExecuteLocal.setJsonEntity(body);
+        return painlessExecuteLocal;
+    }
+
+    private Response performRequestWithRemoteSearchUser(final Request request) throws IOException {
+        request.setOptions(
+            RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", headerFromRandomAuthMethod(REMOTE_SEARCH_USER, PASS))
+        );
+        return client().performRequest(request);
+    }
+}

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

@@ -0,0 +1,303 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.remotecluster;
+
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.util.resource.Resource;
+import org.elasticsearch.test.junit.RunnableTestRuleAdapter;
+import org.junit.ClassRule;
+import org.junit.rules.RuleChain;
+import org.junit.rules.TestRule;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ * Tests cross-cluster painless/execute API under RCS2.0 security model
+ */
+public class RemoteClusterSecurityRCS2PainlessExecuteIT extends AbstractRemoteClusterSecurityTestCase {
+
+    private static final AtomicReference<Map<String, Object>> API_KEY_MAP_REF = new AtomicReference<>();
+    private static final AtomicReference<Map<String, Object>> REST_API_KEY_MAP_REF = new AtomicReference<>();
+    private static final AtomicBoolean SSL_ENABLED_REF = new AtomicBoolean();
+    private static final AtomicBoolean NODE1_RCS_SERVER_ENABLED = new AtomicBoolean();
+    private static final AtomicBoolean NODE2_RCS_SERVER_ENABLED = new AtomicBoolean();
+    private static final AtomicInteger INVALID_SECRET_LENGTH = new AtomicInteger();
+
+    static {
+        fulfillingCluster = ElasticsearchCluster.local()
+            .name("fulfilling-cluster")
+            .nodes(3)
+            .apply(commonClusterConfig)
+            .setting("remote_cluster.port", "0")
+            .setting("xpack.security.remote_cluster_server.ssl.enabled", () -> String.valueOf(SSL_ENABLED_REF.get()))
+            .setting("xpack.security.remote_cluster_server.ssl.key", "remote-cluster.key")
+            .setting("xpack.security.remote_cluster_server.ssl.certificate", "remote-cluster.crt")
+            .setting("xpack.security.authc.token.enabled", "true")
+            .keystore("xpack.security.remote_cluster_server.ssl.secure_key_passphrase", "remote-cluster-password")
+            .node(0, spec -> spec.setting("remote_cluster_server.enabled", "true"))
+            .node(1, spec -> spec.setting("remote_cluster_server.enabled", () -> String.valueOf(NODE1_RCS_SERVER_ENABLED.get())))
+            .node(2, spec -> spec.setting("remote_cluster_server.enabled", () -> String.valueOf(NODE2_RCS_SERVER_ENABLED.get())))
+            .build();
+
+        queryCluster = ElasticsearchCluster.local()
+            .name("query-cluster")
+            .apply(commonClusterConfig)
+            .setting("xpack.security.remote_cluster_client.ssl.enabled", () -> String.valueOf(SSL_ENABLED_REF.get()))
+            .setting("xpack.security.remote_cluster_client.ssl.certificate_authorities", "remote-cluster-ca.crt")
+            .setting("xpack.security.authc.token.enabled", "true")
+            .keystore("cluster.remote.my_remote_cluster.credentials", () -> {
+                if (API_KEY_MAP_REF.get() == null) {
+                    final Map<String, Object> apiKeyMap = createCrossClusterAccessApiKey("""
+                        {
+                          "search": [
+                            {
+                                "names": ["index*"]
+                            }
+                          ]
+                        }""");
+                    API_KEY_MAP_REF.set(apiKeyMap);
+                }
+                return (String) API_KEY_MAP_REF.get().get("encoded");
+            })
+            // Define a bogus API key for another remote cluster
+            .keystore("cluster.remote.invalid_remote.credentials", randomEncodedApiKey())
+            // Define remote with a REST API key to observe expected failure
+            .keystore("cluster.remote.wrong_api_key_type.credentials", () -> {
+                if (REST_API_KEY_MAP_REF.get() == null) {
+                    initFulfillingClusterClient();
+                    final var createApiKeyRequest = new Request("POST", "/_security/api_key");
+                    createApiKeyRequest.setJsonEntity("""
+                        {
+                          "name": "rest_api_key"
+                        }""");
+                    try {
+                        final Response createApiKeyResponse = performRequestWithAdminUser(fulfillingClusterClient, createApiKeyRequest);
+                        assertOK(createApiKeyResponse);
+                        REST_API_KEY_MAP_REF.set(responseAsMap(createApiKeyResponse));
+                    } catch (IOException e) {
+                        throw new UncheckedIOException(e);
+                    }
+                }
+                return (String) REST_API_KEY_MAP_REF.get().get("encoded");
+            })
+            // Define a remote with invalid API key secret length
+            .keystore(
+                "cluster.remote.invalid_secret_length.credentials",
+                () -> Base64.getEncoder()
+                    .encodeToString(
+                        (UUIDs.base64UUID() + ":" + randomAlphaOfLength(INVALID_SECRET_LENGTH.get())).getBytes(StandardCharsets.UTF_8)
+                    )
+            )
+            .rolesFile(Resource.fromClasspath("roles.yml"))
+            .user(REMOTE_METRIC_USER, PASS.toString(), "read_remote_shared_metrics", false)
+            .build();
+    }
+
+    @ClassRule
+    // Use a RuleChain to ensure that fulfilling cluster is started before query cluster
+    // `SSL_ENABLED_REF` is used to control the SSL-enabled setting on the test clusters
+    // We set it here, since randomization methods are not available in the static initialize context above
+    public static TestRule clusterRule = RuleChain.outerRule(new RunnableTestRuleAdapter(() -> {
+        SSL_ENABLED_REF.set(usually());
+        NODE1_RCS_SERVER_ENABLED.set(randomBoolean());
+        NODE2_RCS_SERVER_ENABLED.set(randomBoolean());
+        INVALID_SECRET_LENGTH.set(randomValueOtherThan(22, () -> randomIntBetween(0, 99)));
+    })).around(fulfillingCluster).around(queryCluster);
+
+    @SuppressWarnings({ "unchecked", "checkstyle:LineLength" })
+    public void testPainlessExecute() throws Exception {
+        configureRemoteCluster();
+
+        {
+            // Query cluster -> add role for test user - do not give any privileges for remote_indices
+            final var putRoleRequest = new Request("PUT", "/_security/role/" + REMOTE_SEARCH_ROLE);
+            putRoleRequest.setJsonEntity("""
+                {
+                  "indices": [
+                    {
+                      "names": ["local_index", "my_local*"],
+                      "privileges": ["read"]
+                    }
+                  ]
+                }""");
+            assertOK(adminClient().performRequest(putRoleRequest));
+
+            // Query cluster -> create user and assign role
+            final var putUserRequest = new Request("PUT", "/_security/user/" + REMOTE_SEARCH_USER);
+            putUserRequest.setJsonEntity("""
+                {
+                  "password": "x-pack-test-password",
+                  "roles" : ["remote_search"]
+                }""");
+            assertOK(adminClient().performRequest(putUserRequest));
+
+            // Query cluster -> create test index
+            final var indexDocRequest = new Request("POST", "/local_index/_doc?refresh=true");
+            indexDocRequest.setJsonEntity("{\"local_foo\": \"local_bar\"}");
+            assertOK(client().performRequest(indexDocRequest));
+
+            // Fulfilling cluster -> create test indices
+            final Request bulkRequest = new Request("POST", "/_bulk?refresh=true");
+            bulkRequest.setJsonEntity(Strings.format("""
+                { "index": { "_index": "index1" } }
+                { "foo": "bar" }
+                { "index": { "_index": "secretindex" } }
+                { "bar": "foo" }
+                """));
+            assertOK(performRequestAgainstFulfillingCluster(bulkRequest));
+        }
+
+        {
+            // TEST CASE 1: Query local cluster for local_index - should work since role has read perms for it
+            Request painlessExecuteLocal = createPainlessExecuteRequest("local_index");
+
+            Response response = performRequestWithRemoteSearchUser(painlessExecuteLocal);
+            assertOK(response);
+            String responseBody = EntityUtils.toString(response.getEntity());
+            assertThat(responseBody, equalTo("{\"result\":[\"test\"]}"));
+        }
+        {
+            // update role to have permissions to remote index* pattern
+            var updateRoleRequest = new Request("PUT", "/_security/role/" + REMOTE_SEARCH_ROLE);
+            updateRoleRequest.setJsonEntity("""
+                {
+                  "indices": [
+                    {
+                      "names": ["local_index", "my_local*"],
+                      "privileges": ["read"]
+                    }
+                  ],
+                  "remote_indices": [
+                    {
+                      "names": ["index*"],
+                      "privileges": ["read", "read_cross_cluster"],
+                      "clusters": ["my_remote_cluster"]
+                    }
+                  ]
+                }""");
+
+            assertOK(adminClient().performRequest(updateRoleRequest));
+        }
+        {
+            // TEST CASE 2: Query remote cluster for secretindex - should fail since no perms granted for it
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:secretindex");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [secretindex]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // TEST CASE 3: Query remote cluster for index1 - should succeed since read and cross-cluster-read perms granted
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index1");
+            Response response = performRequestWithRemoteSearchUser(painlessExecuteRemote);
+            String responseBody = EntityUtils.toString(response.getEntity());
+            assertOK(response);
+            assertThat(responseBody, equalTo("{\"result\":[\"test\"]}"));
+        }
+        {
+            // TEST CASE 4: Query local cluster for not_present index - should fail with 403 since role does not have perms for this index
+            Request painlessExecuteLocal = createPainlessExecuteRequest("index_not_present");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteLocal));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [index_not_present]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // TEST CASE 5: Query local cluster for my_local_123 index - role has perms for this pattern, but index does not exist, so 404
+            Request painlessExecuteLocal = createPainlessExecuteRequest("my_local_123");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteLocal));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(404));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("\"type\":\"index_not_found_exception\""));
+        }
+        {
+            // TEST CASE 6: Query local cluster for my_local* index - painless/execute does not allow wildcards, so fails with 400
+            Request painlessExecuteLocal = createPainlessExecuteRequest("my_local*");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteLocal));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(400));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("indices:data/read/scripts/painless/execute does not support wildcards"));
+            assertThat(errorResponseBody, containsString("\"type\":\"illegal_argument_exception\""));
+        }
+        {
+            // TEST CASE 7: Query remote cluster for cluster that does not exist, and user does not have perms for that pattern - 403 ???
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:abc123");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(403));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("unauthorized for user [remote_search_user]"));
+            assertThat(errorResponseBody, containsString("on indices [abc123]"));
+            assertThat(errorResponseBody, containsString("\"type\":\"security_exception\""));
+        }
+        {
+            // TEST CASE 8: Query remote cluster for cluster that does not exist, but has permissions for the index pattern - 404
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index123");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(404));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("\"type\":\"index_not_found_exception\""));
+        }
+        {
+            // TEST CASE 9: Query remote cluster with wildcard in index - painless/execute does not allow wildcards, so fails with 400
+            Request painlessExecuteRemote = createPainlessExecuteRequest("my_remote_cluster:index*");
+            ResponseException exc = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(painlessExecuteRemote));
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), is(400));
+            String errorResponseBody = EntityUtils.toString(exc.getResponse().getEntity());
+            assertThat(errorResponseBody, containsString("indices:data/read/scripts/painless/execute does not support wildcards"));
+            assertThat(errorResponseBody, containsString("\"type\":\"illegal_argument_exception\""));
+        }
+    }
+
+    private static Request createPainlessExecuteRequest(String indexExpression) {
+        Request painlessExecuteLocal = new Request("POST", "_scripts/painless/_execute");
+        String body = """
+            {
+                "script": {
+                    "source": "emit(\\"test\\")"
+                },
+                "context": "keyword_field",
+                "context_setup": {
+                    "index": "INDEX_EXPRESSION_HERE",
+                    "document": {
+                        "@timestamp": "2023-05-06T16:22:22.000Z"
+                    }
+                }
+            }""".replace("INDEX_EXPRESSION_HERE", indexExpression);
+        painlessExecuteLocal.setJsonEntity(body);
+        return painlessExecuteLocal;
+    }
+
+    private Response performRequestWithRemoteSearchUser(final Request request) throws IOException {
+        request.setOptions(
+            RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", headerFromRandomAuthMethod(REMOTE_SEARCH_USER, PASS))
+        );
+        return client().performRequest(request);
+    }
+}

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

@@ -175,7 +175,6 @@ public class RemoteClusterSecurityResolveClusterIT extends AbstractRemoteCluster
             Response response = performRequestWithRemoteSearchUser(starResolveRequest);
             assertOK(response);
             Map<String, Object> responseMap = responseAsMap(response);
-            System.err.println(">> XXX CASE1 remoteClusterResponse: " + responseMap);
             assertLocalMatching(responseMap);
 
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java

@@ -477,7 +477,7 @@ public class AuthorizationService {
                     resolvedIndicesListener.onResponse(resolvedIndices);
                     return;
                 }
-                final ResolvedIndices resolvedIndices = IndicesAndAliasesResolver.tryResolveWithoutWildcards(action, request);
+                final ResolvedIndices resolvedIndices = indicesAndAliasesResolver.tryResolveWithoutWildcards(action, request);
                 if (resolvedIndices != null) {
                     resolvedIndicesListener.onResponse(resolvedIndices);
                 } else {
@@ -870,8 +870,8 @@ public class AuthorizationService {
         }, listener::onFailure));
     }
 
-    private static String resolveIndexNameDateMath(BulkItemRequest bulkItemRequest) {
-        final ResolvedIndices resolvedIndices = IndicesAndAliasesResolver.resolveIndicesAndAliasesWithoutWildcards(
+    private String resolveIndexNameDateMath(BulkItemRequest bulkItemRequest) {
+        final ResolvedIndices resolvedIndices = indicesAndAliasesResolver.resolveIndicesAndAliasesWithoutWildcards(
             getAction(bulkItemRequest),
             bulkItemRequest.request()
         );

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolver.java

@@ -46,7 +46,6 @@ import java.util.SortedMap;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
-import java.util.stream.Stream;
 
 import static org.elasticsearch.xpack.core.security.authz.IndicesAndAliasesResolverField.NO_INDEX_PLACEHOLDER;
 
@@ -120,7 +119,7 @@ class IndicesAndAliasesResolver {
      * @return The {@link ResolvedIndices} or null if wildcard expansion must be performed.
      */
     @Nullable
-    static ResolvedIndices tryResolveWithoutWildcards(String action, TransportRequest transportRequest) {
+    ResolvedIndices tryResolveWithoutWildcards(String action, TransportRequest transportRequest) {
         // We only take care of IndicesRequest
         if (false == transportRequest instanceof IndicesRequest) {
             return null;
@@ -145,7 +144,7 @@ class IndicesAndAliasesResolver {
         return false;
     }
 
-    static ResolvedIndices resolveIndicesAndAliasesWithoutWildcards(String action, IndicesRequest indicesRequest) {
+    ResolvedIndices resolveIndicesAndAliasesWithoutWildcards(String action, IndicesRequest indicesRequest) {
         assert false == requiresWildcardExpansion(indicesRequest) : "request must not require wildcard expansion";
         final String[] indices = indicesRequest.indices();
         if (indices == null || indices.length == 0) {
@@ -162,20 +161,28 @@ class IndicesAndAliasesResolver {
             );
         }
 
+        final ResolvedIndices split;
+        if (indicesRequest instanceof IndicesRequest.SingleIndexNoWildcards single && single.allowsRemoteIndices()) {
+            split = remoteClusterResolver.splitLocalAndRemoteIndexNames(indicesRequest.indices());
+        } else {
+            split = new ResolvedIndices(Arrays.asList(indicesRequest.indices()), List.of());
+        }
+
         // NOTE: shard level requests do support wildcards (as they hold the original indices options) but don't support
         // replacing their indices.
         // That is fine though because they never contain wildcards, as they get replaced as part of the authorization of their
         // corresponding parent request on the coordinating node. Hence wildcards don't need to get replaced nor exploded for
         // shard level requests.
-        final List<String> localIndices = new ArrayList<>(indices.length);
-        for (String name : indices) {
+        final List<String> localIndices = new ArrayList<>(split.getLocal().size());
+        for (String localName : split.getLocal()) {
             // TODO: Shard level requests have wildcard expanded already and do not need go through this check
-            if (Regex.isSimpleMatchPattern(name)) {
-                throwOnUnexpectedWildcards(action, indices);
+            if (Regex.isSimpleMatchPattern(localName)) {
+                throwOnUnexpectedWildcards(action, split.getLocal());
             }
-            localIndices.add(IndexNameExpressionResolver.resolveDateMathExpression(name));
+            localIndices.add(IndexNameExpressionResolver.resolveDateMathExpression(localName));
         }
-        return new ResolvedIndices(localIndices, List.of());
+
+        return new ResolvedIndices(localIndices, split.getRemote());
     }
 
     /**
@@ -196,8 +203,8 @@ class IndicesAndAliasesResolver {
         return split;
     }
 
-    private static void throwOnUnexpectedWildcards(String action, String[] indices) {
-        final List<String> wildcards = Stream.of(indices).filter(Regex::isSimpleMatchPattern).toList();
+    private static void throwOnUnexpectedWildcards(String action, List<String> indices) {
+        final List<String> wildcards = indices.stream().filter(Regex::isSimpleMatchPattern).toList();
         assert wildcards.isEmpty() == false : "we already know that there's at least one wildcard in the indices";
         throw new IllegalArgumentException(
             "the action "

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/RBACEngine.java

@@ -408,7 +408,12 @@ public class RBACEngine implements AuthorizationEngine {
     }
 
     private static boolean allowsRemoteIndices(TransportRequest transportRequest) {
-        return transportRequest instanceof IndicesRequest.Replaceable replaceable && replaceable.allowsRemoteIndices();
+        // TODO this may need to change. See https://github.com/elastic/elasticsearch/issues/105598
+        if (transportRequest instanceof IndicesRequest.SingleIndexNoWildcards single) {
+            return single.allowsRemoteIndices();
+        } else {
+            return transportRequest instanceof IndicesRequest.Replaceable replaceable && replaceable.allowsRemoteIndices();
+        }
     }
 
     private static boolean isChildActionAuthorizedByParentOnLocalNode(RequestInfo requestInfo, AuthorizationInfo authorizationInfo) {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolverTests.java

@@ -404,7 +404,7 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         // aliases with names starting with '-' or '+' can be created up to version 5.x and can be around in 6.x
         ShardSearchRequest request = mock(ShardSearchRequest.class);
         when(request.indices()).thenReturn(new String[] { "-index10", "-index20", "+index30" });
-        List<String> indices = IndicesAndAliasesResolver.resolveIndicesAndAliasesWithoutWildcards(
+        List<String> indices = defaultIndicesResolver.resolveIndicesAndAliasesWithoutWildcards(
             TransportSearchAction.TYPE.name() + "[s]",
             request
         ).getLocal();
@@ -418,7 +418,7 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         when(request.indices()).thenReturn(new String[] { "index*" });
         IllegalArgumentException exception = expectThrows(
             IllegalArgumentException.class,
-            () -> IndicesAndAliasesResolver.resolveIndicesAndAliasesWithoutWildcards(TransportSearchAction.TYPE.name() + "[s]", request)
+            () -> defaultIndicesResolver.resolveIndicesAndAliasesWithoutWildcards(TransportSearchAction.TYPE.name() + "[s]", request)
         );
         assertThat(
             exception,
@@ -443,7 +443,7 @@ public class IndicesAndAliasesResolverTests extends ESTestCase {
         }
         IllegalArgumentException exception = expectThrows(
             IllegalArgumentException.class,
-            () -> IndicesAndAliasesResolver.resolveIndicesAndAliasesWithoutWildcards(TransportSearchAction.TYPE.name() + "[s]", request)
+            () -> defaultIndicesResolver.resolveIndicesAndAliasesWithoutWildcards(TransportSearchAction.TYPE.name() + "[s]", request)
         );
 
         assertThat(