Limit headers stored by reindexing to security (#49267)

This is related to #42612. It adds a setting to configure what headers are
stored by the persistent reindexing task for further requests. Additionally,
it has the x-pack security module automatically configure this setting to
ensure security works with reindexing.

Author: Tim-Brooks

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportStartReindexTaskAction.java

@@ -29,8 +29,10 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.persistent.PersistentTasksService;
@@ -39,11 +41,15 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.List;
+import java.util.Map;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 public class TransportStartReindexTaskAction
     extends HandledTransportAction<StartReindexTaskAction.Request, StartReindexTaskAction.Response> {
 
+    private final List<String> headersToInclude;
     private final ThreadPool threadPool;
     private final PersistentTasksService persistentTasksService;
     private final ReindexValidator reindexValidator;
@@ -55,6 +61,7 @@ public TransportStartReindexTaskAction(Settings settings, Client client, Transpo
                                            ClusterService clusterService, PersistentTasksService persistentTasksService,
                                            AutoCreateIndex autoCreateIndex, NamedXContentRegistry xContentRegistry) {
         super(StartReindexTaskAction.NAME, transportService, actionFilters, StartReindexTaskAction.Request::new);
+        this.headersToInclude = ReindexHeaders.REINDEX_INCLUDED_HEADERS.get(settings);
         this.threadPool = threadPool;
         this.reindexValidator = new ReindexValidator(settings, clusterService, indexNameExpressionResolver, autoCreateIndex);
         this.persistentTasksService = persistentTasksService;
@@ -72,9 +79,15 @@ protected void doExecute(Task task, StartReindexTaskAction.Request request, Acti
 
         String generatedId = UUIDs.randomBase64UUID();
 
+        ThreadContext threadContext = threadPool.getThreadContext();
+        Map<String, String> included = headersToInclude.stream()
+            .map(header -> new Tuple<>(header, threadContext.getHeader(header)))
+            .filter(t -> t.v2() != null)
+            .collect(Collectors.toMap(Tuple::v1, Tuple::v2));
+
         // In the current implementation, we only need to store task results if we do not wait for completion
         boolean storeTaskResult = request.getWaitForCompletion() == false;
-        ReindexTaskParams job = new ReindexTaskParams(storeTaskResult, threadPool.getThreadContext().getHeaders());
+        ReindexTaskParams job = new ReindexTaskParams(storeTaskResult, included);
 
         ReindexTaskStateDoc reindexState = new ReindexTaskStateDoc(request.getReindexRequest());
         reindexIndexClient.createReindexTaskDoc(generatedId, reindexState, new ActionListener<>() {

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -75,6 +75,7 @@
 import org.elasticsearch.http.HttpTransportSettings;
 import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.reindex.ReindexHeaders;
 import org.elasticsearch.indices.IndexingMemoryController;
 import org.elasticsearch.indices.IndicesQueryCache;
 import org.elasticsearch.indices.IndicesRequestCache;
@@ -462,7 +463,8 @@ public void apply(Settings value, Settings current, Settings previous) {
             TransportAddVotingConfigExclusionsAction.MAXIMUM_VOTING_CONFIG_EXCLUSIONS_SETTING,
             ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING,
             ClusterBootstrapService.UNCONFIGURED_BOOTSTRAP_TIMEOUT_SETTING,
-            LagDetector.CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING);
+            LagDetector.CLUSTER_FOLLOWER_LAG_TIMEOUT_SETTING,
+            ReindexHeaders.REINDEX_INCLUDED_HEADERS);
 
     static List<SettingUpgrader<?>> BUILT_IN_SETTING_UPGRADERS = Collections.emptyList();
 

server/src/main/java/org/elasticsearch/index/reindex/ReindexHeaders.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.reindex;
+
+import org.elasticsearch.common.settings.Setting;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+public class ReindexHeaders {
+
+    /**
+     * A list of headers that should be extracted from the start reindex request and used on subsequent
+     * requests when resilient reindexing is enabled. For example, any authorization headers required for
+     * reindexing should be configured.
+     */
+    public static final Setting<List<String>> REINDEX_INCLUDED_HEADERS = Setting.listSetting("reindex.request_headers.include",
+        Collections.emptyList(), Function.identity(), Setting.Property.NodeScope);
+}

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

@@ -42,6 +42,7 @@
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.index.IndexModule;
+import org.elasticsearch.index.reindex.ReindexHeaders;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.license.License;
@@ -112,6 +113,7 @@
 import org.elasticsearch.xpack.core.security.action.user.PutUserAction;
 import org.elasticsearch.xpack.core.security.action.user.SetEnabledAction;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationFailureHandler;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationServiceField;
 import org.elasticsearch.xpack.core.security.authc.DefaultAuthenticationFailureHandler;
 import org.elasticsearch.xpack.core.security.authc.InternalRealmsSettings;
@@ -568,6 +570,8 @@ static Settings additionalSettings(final Settings settings, final boolean enable
                 SecurityHttpSettings.overrideSettings(builder, settings);
             }
             builder.put(SecuritySettings.addUserSettings(settings));
+            List<String> authHeaders = Arrays.asList(AuthenticationField.AUTHENTICATION_KEY, AuthenticationServiceField.RUN_AS_USER_HEADER);
+            builder.put(ReindexHeaders.REINDEX_INCLUDED_HEADERS.getKey(), Strings.collectionToCommaDelimitedString(authHeaders));
             return builder.build();
         } else {
             return Settings.EMPTY;