Adjust interception of requests for specific shard IDs (#101656)

Some index requests target shard IDs specifically, which may not match the indices that the request targets as given by `IndicesRequest#indices()`, which requires a different interception strategy in order to make sure those requests are handled correctly in all cases and that any malformed messages are caught early to aid in troubleshooting.

This PR adds and interface allowing requests to report the shard IDs they target as well as the index names, and adjusts the interception of those requests as appropriate to handle those shard IDs in the cases where they are relevant.

Author: gwbrown

docs/changelog/101656.yaml

@@ -0,0 +1,5 @@
+pr: 101656
+summary: Adjust interception of requests for specific shard IDs
+area: Authorization
+type: bug
+issues: []

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

@@ -9,6 +9,9 @@
 package org.elasticsearch.action;
 
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.index.shard.ShardId;
+
+import java.util.Collection;
 
 /**
  * Needs to be implemented by all {@link org.elasticsearch.action.ActionRequest} subclasses that relate to
@@ -72,4 +75,19 @@ default boolean allowsRemoteIndices() {
             return true;
         }
     }
+
+    /**
+     * This subtype of request is for requests which may travel to remote clusters. These requests may need to provide additional
+     * information to the system on top of the indices the action relates to in order to be handled correctly in all cases.
+     */
+    interface RemoteClusterShardRequest extends IndicesRequest {
+        /**
+         * Returns the shards this action is targeting directly, which may not obviously align with the indices returned by
+         * {@code indices()}. This is mostly used by requests which fan out to a number of shards for the those fan-out requests.
+         *
+         * A default is intentionally not provided for this method. It is critical that this method be implemented correctly for all
+         * remote cluster requests,
+         */
+        Collection<ShardId> shards();
+    }
 }

server/src/main/java/org/elasticsearch/action/get/GetRequest.java

@@ -64,7 +64,7 @@ public class GetRequest extends SingleShardRequest<GetRequest> implements Realti
 
     public GetRequest() {}
 
-    GetRequest(StreamInput in) throws IOException {
+    public GetRequest(StreamInput in) throws IOException {
         super(in);
         if (in.getTransportVersion().before(TransportVersions.V_8_0_0)) {
             in.readString();

server/src/main/java/org/elasticsearch/action/get/GetResponse.java

@@ -34,7 +34,7 @@ public class GetResponse extends ActionResponse implements Iterable<DocumentFiel
 
     GetResult getResult;
 
-    GetResponse(StreamInput in) throws IOException {
+    public GetResponse(StreamInput in) throws IOException {
         super(in);
         getResult = new GetResult(in);
     }

server/src/main/java/org/elasticsearch/action/support/single/shard/SingleShardRequest.java

@@ -19,8 +19,12 @@
 import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 
-public abstract class SingleShardRequest<Request extends SingleShardRequest<Request>> extends ActionRequest implements IndicesRequest {
+public abstract class SingleShardRequest<Request extends SingleShardRequest<Request>> extends ActionRequest
+    implements
+        IndicesRequest.RemoteClusterShardRequest {
 
     public static final IndicesOptions INDICES_OPTIONS = IndicesOptions.strictSingleIndexNoExpandForbidClosed();
 
@@ -85,6 +89,11 @@ public String[] indices() {
         return new String[] { index };
     }
 
+    @Override
+    public List<ShardId> shards() {
+        return Collections.singletonList(this.internalShardId);
+    }
+
     @Override
     public IndicesOptions indicesOptions() {
         return INDICES_OPTIONS;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/downsample/DownsampleShardTask.java

@@ -98,6 +98,10 @@ public DownsampleConfig config() {
         return config;
     }
 
+    public ShardId shardId() {
+        return shardId;
+    }
+
     public long getTotalShardDocCount() {
         return totalShardDocCount;
     }

x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/DownsampleShardPersistentTaskExecutor.java

@@ -49,6 +49,7 @@
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.Executor;
@@ -260,7 +261,7 @@ private DelegatingAction() {
             super(NAME);
         }
 
-        public static class Request extends ActionRequest implements IndicesRequest {
+        public static class Request extends ActionRequest implements IndicesRequest.RemoteClusterShardRequest {
 
             private final DownsampleShardTask task;
             private final BytesRef lastDownsampleTsid;
@@ -291,6 +292,11 @@ public IndicesOptions indicesOptions() {
             public void writeTo(StreamOutput out) {
                 throw new IllegalStateException("request should stay local");
             }
+
+            @Override
+            public Collection<ShardId> shards() {
+                return Collections.singletonList(task.shardId());
+            }
         }
 
         public static class TA extends TransportAction<Request, ActionResponse.Empty> {

x-pack/plugin/security/qa/consistency-checks/build.gradle

@@ -0,0 +1,26 @@
+apply plugin: 'elasticsearch.standalone-test'
+
+dependencies {
+
+  testImplementation(testArtifact(project(xpackModule('core'))))
+  testImplementation project(path: ':modules:ingest-common')
+  testImplementation project(path: ':modules:data-streams')
+  testImplementation project(path: ':modules:lang-mustache')
+  testImplementation project(path: ':modules:rank-eval')
+  testImplementation project(path: ':modules:reindex')
+  testImplementation project(path: xpackModule('analytics'))
+  testImplementation project(path: xpackModule('async-search'))
+  testImplementation project(path: xpackModule('autoscaling'))
+  testImplementation project(path: xpackModule('ccr'))
+  testImplementation project(path: xpackModule('downsample'))
+  testImplementation project(path: xpackModule('eql'))
+  testImplementation project(path: xpackModule('esql'))
+  testImplementation project(path: xpackModule('frozen-indices'))
+  testImplementation project(path: xpackModule('graph'))
+  testImplementation project(path: xpackModule('ilm'))
+  testImplementation project(path: xpackModule('inference'))
+  testImplementation project(path: xpackModule('profiling'))
+  testImplementation project(path: xpackModule('rollup'))
+  testImplementation project(path: xpackModule('slm'))
+  testImplementation project(path: xpackModule('sql'))
+}

x-pack/plugin/security/qa/consistency-checks/src/test/java/org/elasticsearch/xpack/security/CrossClusterShardTests.java

@@ -0,0 +1,165 @@
+/*
+ * 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.security;
+
+import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsAction;
+import org.elasticsearch.action.search.TransportSearchShardsAction;
+import org.elasticsearch.action.support.TransportAction;
+import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
+import org.elasticsearch.common.inject.Binding;
+import org.elasticsearch.common.inject.TypeLiteral;
+import org.elasticsearch.datastreams.DataStreamsPlugin;
+import org.elasticsearch.index.rankeval.RankEvalPlugin;
+import org.elasticsearch.ingest.IngestTestPlugin;
+import org.elasticsearch.ingest.common.IngestCommonPlugin;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.reindex.ReindexPlugin;
+import org.elasticsearch.script.mustache.MustachePlugin;
+import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.xpack.analytics.AnalyticsPlugin;
+import org.elasticsearch.xpack.autoscaling.Autoscaling;
+import org.elasticsearch.xpack.ccr.Ccr;
+import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
+import org.elasticsearch.xpack.core.security.action.apikey.CrossClusterApiKeyRoleDescriptorBuilder;
+import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
+import org.elasticsearch.xpack.downsample.Downsample;
+import org.elasticsearch.xpack.downsample.DownsampleShardPersistentTaskExecutor;
+import org.elasticsearch.xpack.eql.plugin.EqlPlugin;
+import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
+import org.elasticsearch.xpack.frozen.FrozenIndices;
+import org.elasticsearch.xpack.graph.Graph;
+import org.elasticsearch.xpack.ilm.IndexLifecycle;
+import org.elasticsearch.xpack.inference.InferencePlugin;
+import org.elasticsearch.xpack.profiling.ProfilingPlugin;
+import org.elasticsearch.xpack.rollup.Rollup;
+import org.elasticsearch.xpack.search.AsyncSearch;
+import org.elasticsearch.xpack.slm.SnapshotLifecycle;
+import org.elasticsearch.xpack.sql.plugin.SqlPlugin;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.function.Predicate;
+
+public class CrossClusterShardTests extends ESSingleNodeTestCase {
+
+    Set<String> MANUALLY_CHECKED_SHARD_ACTIONS = Set.of(
+        // The request types for these actions are all subtypes of SingleShardRequest, and have been evaluated to make sure their
+        // `shards()` methods return the correct thing.
+        TransportSearchShardsAction.NAME,
+
+        // These types have had the interface implemented manually.
+        DownsampleShardPersistentTaskExecutor.DelegatingAction.NAME,
+
+        // These actions do not have any references to shard IDs in their requests.
+        ClusterSearchShardsAction.NAME
+    );
+
+    Set<Class<?>> CHECKED_ABSTRACT_CLASSES = Set.of(
+        // This abstract class implements the interface so we can assume all of its subtypes do so properly as well.
+        TransportSingleShardAction.class
+    );
+
+    @Override
+    protected Collection<Class<? extends Plugin>> getPlugins() {
+        final ArrayList<Class<? extends Plugin>> plugins = new ArrayList<>(super.getPlugins());
+        plugins.addAll(
+            List.of(
+                LocalStateCompositeXPackPlugin.class,
+                AnalyticsPlugin.class,
+                AsyncSearch.class,
+                Autoscaling.class,
+                Ccr.class,
+                DataStreamsPlugin.class,
+                Downsample.class,
+                EqlPlugin.class,
+                EsqlPlugin.class,
+                FrozenIndices.class,
+                Graph.class,
+                IndexLifecycle.class,
+                InferencePlugin.class,
+                IngestCommonPlugin.class,
+                IngestTestPlugin.class,
+                MustachePlugin.class,
+                ProfilingPlugin.class,
+                RankEvalPlugin.class,
+                ReindexPlugin.class,
+                Rollup.class,
+                SnapshotLifecycle.class,
+                SqlPlugin.class
+            )
+        );
+        return plugins;
+    }
+
+    @SuppressWarnings("rawtypes")
+    public void testCheckForNewShardLevelTransportActions() throws Exception {
+        Node node = node();
+        List<Binding<TransportAction>> transportActionBindings = node.injector().findBindingsByType(TypeLiteral.get(TransportAction.class));
+        Set<String> crossClusterPrivilegeNames = new HashSet<>();
+        crossClusterPrivilegeNames.addAll(List.of(CrossClusterApiKeyRoleDescriptorBuilder.CCS_INDICES_PRIVILEGE_NAMES));
+        crossClusterPrivilegeNames.addAll(List.of(CrossClusterApiKeyRoleDescriptorBuilder.CCR_INDICES_PRIVILEGE_NAMES));
+
+        List<String> shardActions = transportActionBindings.stream()
+            .map(binding -> binding.getProvider().get())
+            .filter(action -> IndexPrivilege.get(crossClusterPrivilegeNames).predicate().test(action.actionName))
+            .filter(this::actionIsLikelyShardAction)
+            .map(action -> action.actionName)
+            .toList();
+
+        List<String> actionsNotOnAllowlist = shardActions.stream().filter(Predicate.not(MANUALLY_CHECKED_SHARD_ACTIONS::contains)).toList();
+        if (actionsNotOnAllowlist.isEmpty() == false) {
+            fail("""
+                If this test fails, you likely just added a transport action, probably with `shard` in the name. Transport actions which
+                operate on shards directly and can be used across clusters must meet some additional requirements in order to be
+                handled correctly by all Elasticsearch infrastructure, so please make sure you have read the javadoc on the
+                IndicesRequest.RemoteClusterShardRequest interface and implemented it if appropriate and not already appropriately
+                implemented by a supertype, then add the name (as in "indices:data/read/get") of your new transport action to
+                MANUALLY_CHECKED_SHARD_ACTIONS above. Found actions not in allowlist:
+                """ + actionsNotOnAllowlist);
+        }
+
+        // Also make sure the allowlist stays up to date and doesn't have any unnecessary entries.
+        List<String> actionsOnAllowlistNotFound = MANUALLY_CHECKED_SHARD_ACTIONS.stream()
+            .filter(Predicate.not(shardActions::contains))
+            .toList();
+        if (actionsOnAllowlistNotFound.isEmpty() == false) {
+            fail(
+                "Some actions were on the allowlist but not found in the list of cross-cluster capable transport actions, please remove "
+                    + "these from MANUALLY_CHECKED_SHARD_ACTIONS if they have been removed from Elasticsearch: "
+                    + actionsOnAllowlistNotFound
+            );
+        }
+    }
+
+    /**
+     * Getting to the actual request classes themselves is made difficult by the design of Elasticsearch's transport
+     * protocol infrastructure combined with JVM type erasure. Therefore, we resort to a crude heuristic here.
+     * @param transportAction The transportport action to be checked.
+     * @return True if the action is suspected of being an action which may operate on shards directly.
+     */
+    private boolean actionIsLikelyShardAction(TransportAction<?, ?> transportAction) {
+        Class<?> clazz = transportAction.getClass();
+        Set<Class<?>> classHeirarchy = new HashSet<>();
+        while (clazz != TransportAction.class) {
+            classHeirarchy.add(clazz);
+            clazz = clazz.getSuperclass();
+        }
+        boolean hasCheckedSuperclass = classHeirarchy.stream().anyMatch(clz -> CHECKED_ABSTRACT_CLASSES.contains(clz));
+        boolean shardInClassName = classHeirarchy.stream().anyMatch(clz -> clz.getName().toLowerCase(Locale.ROOT).contains("shard"));
+        return hasCheckedSuperclass == false
+            && (shardInClassName
+                || transportAction.actionName.toLowerCase(Locale.ROOT).contains("shard")
+                || transportAction.actionName.toLowerCase(Locale.ROOT).contains("[s]"));
+    }
+
+}