Copy http headers to ThreadContext strictly

modules/reindex/src/test/java/org/elasticsearch/index/reindex/ReindexFromRemoteWithAuthTests.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.RestHeaderDefinition;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.tasks.Task;
@@ -161,8 +162,9 @@ public List<ActionFilter> getActionFilters() {
         }
 
         @Override
-        public Collection<String> getRestHeaders() {
-            return Arrays.asList(TestFilter.AUTHORIZATION_HEADER, TestFilter.EXAMPLE_HEADER);
+        public Collection<RestHeaderDefinition> getRestHeaders() {
+            return Arrays.asList(new RestHeaderDefinition(TestFilter.AUTHORIZATION_HEADER, false),
+                new RestHeaderDefinition(TestFilter.EXAMPLE_HEADER, false));
         }
     }
 

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

@@ -227,6 +227,7 @@
 import org.elasticsearch.plugins.ActionPlugin.ActionHandler;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestHeaderDefinition;
 import org.elasticsearch.rest.action.RestFieldCapabilitiesAction;
 import org.elasticsearch.rest.action.RestMainAction;
 import org.elasticsearch.rest.action.admin.cluster.RestAddVotingConfigExclusionAction;
@@ -384,9 +385,9 @@ public ActionModule(Settings settings, IndexNameExpressionResolver indexNameExpr
         actionFilters = setupActionFilters(actionPlugins);
         autoCreateIndex = new AutoCreateIndex(settings, clusterSettings, indexNameExpressionResolver);
         destructiveOperations = new DestructiveOperations(settings, clusterSettings);
-        Set<String> headers = Stream.concat(
+        Set<RestHeaderDefinition> headers = Stream.concat(
             actionPlugins.stream().flatMap(p -> p.getRestHeaders().stream()),
-            Stream.of(Task.X_OPAQUE_ID)
+            Stream.of(new RestHeaderDefinition(Task.X_OPAQUE_ID, false))
         ).collect(Collectors.toSet());
         UnaryOperator<RestHandler> restWrapper = null;
         for (ActionPlugin plugin : actionPlugins) {

server/src/main/java/org/elasticsearch/plugins/ActionPlugin.java

@@ -36,6 +36,7 @@
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestHeaderDefinition;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -91,7 +92,7 @@ default List<RestHandler> getRestHandlers(Settings settings, RestController rest
     /**
      * Returns headers which should be copied through rest requests on to internal requests.
      */
-    default Collection<String> getRestHeaders() {
+    default Collection<RestHeaderDefinition> getRestHeaders() {
         return Collections.emptyList();
     }
 

server/src/main/java/org/elasticsearch/rest/RestController.java

@@ -71,10 +71,10 @@ public class RestController implements HttpServerTransport.Dispatcher {
     private final CircuitBreakerService circuitBreakerService;
 
     /** Rest headers that are copied to internal requests made during a rest request. */
-    private final Set<String> headersToCopy;
+    private final Set<RestHeaderDefinition> headersToCopy;
     private final UsageService usageService;
 
-    public RestController(Set<String> headersToCopy, UnaryOperator<RestHandler> handlerWrapper,
+    public RestController(Set<RestHeaderDefinition> headersToCopy, UnaryOperator<RestHandler> handlerWrapper,
             NodeClient client, CircuitBreakerService circuitBreakerService, UsageService usageService) {
         this.headersToCopy = headersToCopy;
         this.usageService = usageService;
@@ -255,11 +255,19 @@ private void sendContentTypeErrorMessage(@Nullable List<String> contentTypeHeade
     }
 
     private void tryAllHandlers(final RestRequest request, final RestChannel channel, final ThreadContext threadContext) throws Exception {
-        for (String key : headersToCopy) {
-            String httpHeader = request.header(key);
-            if (httpHeader != null) {
-                threadContext.putHeader(key, httpHeader);
+        try {
+            for (RestHeaderDefinition restHeader : headersToCopy) {
+                final String name = restHeader.getName();
+                String httpHeader = restHeader.isMultiValueAllowed() ? request.getAllHeaderValuesAsString(name) :
+                    request.getSingleValuedHeader(name);
+                if (httpHeader != null) {
+                    threadContext.putHeader(name, httpHeader);
+                }
             }
+        } catch (IllegalStateException e){
+            channel.sendResponse(
+                BytesRestResponse.createSimpleErrorResponse(channel, BAD_REQUEST, e.getMessage()));
+            return;
         }
         // error_trace cannot be used when we disable detailed errors
         // we consume the error_trace parameter first to ensure that it is always consumed

server/src/main/java/org/elasticsearch/rest/RestHeaderDefinition.java

@@ -0,0 +1,46 @@
+/*
+ * 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.rest;
+
+/**
+ * A definition for an http header that should be copied to the {@link org.elasticsearch.common.util.concurrent.ThreadContext} when
+ * reading the request on the rest layer.
+ */
+public final class RestHeaderDefinition {
+    private final String name;
+    /**
+     * This should be set to true only when the syntax of the value of the Header to copy is defined as a comma separated list of String
+     * values.
+     */
+    private final boolean multiValueAllowed;
+
+    public RestHeaderDefinition(String name, boolean multiValueAllowed) {
+        this.name = name;
+        this.multiValueAllowed = multiValueAllowed;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public boolean isMultiValueAllowed() {
+        return multiValueAllowed;
+    }
+}

server/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -233,6 +233,32 @@ public final List<String> getAllHeaderValues(String name) {
         return null;
     }
 
+    /**
+     * Get all distinct values for the header as a comma separated String or {@code null} if the header was not found
+     */
+    public final String getAllHeaderValuesAsString(String name) {
+        List<String> values = headers.get(name);
+        if (values != null) {
+            return values.stream().distinct().collect(Collectors.joining(","));
+        }
+        return null;
+    }
+
+    /**
+     * Get the value from a header that is expected to be single valued. Return null if the header is not found or throw an exception if it
+     * has multiple values.
+     */
+    public final String getSingleValuedHeader(String name) {
+        List<String> values = headers.get(name);
+        if (values != null && values.isEmpty() == false) {
+            if (values.size() > 1 && new HashSet<>(values).size() > 1) {
+                throw new IllegalStateException("multiple values for single-valued header [" + name + "].");
+            }
+            return values.get(0);
+        }
+        return null;
+    }
+
     /**
      * Get all of the headers and values associated with the headers. Modifications of this map are not supported.
      */

server/src/test/java/org/elasticsearch/rest/RestControllerTests.java

@@ -104,7 +104,8 @@ public void setup() {
 
     public void testApplyRelevantHeaders() throws Exception {
         final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
-        Set<String> headers = new HashSet<>(Arrays.asList("header.1", "header.2"));
+        Set<RestHeaderDefinition> headers = new HashSet<>(Arrays.asList(new RestHeaderDefinition("header.1", true),
+            new RestHeaderDefinition("header.2", true)));
         final RestController restController = new RestController(headers, null, null, circuitBreakerService, usageService);
         Map<String, List<String>> restHeaders = new HashMap<>();
         restHeaders.put("header.1", Collections.singletonList("true"));
@@ -137,6 +138,40 @@ public MethodHandlers next() {
         assertNull(threadContext.getHeader("header.3"));
     }
 
+    public void testRequestWithDisallowedMultiValuedHeader() {
+        final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        Set<RestHeaderDefinition> headers = new HashSet<>(Arrays.asList(new RestHeaderDefinition("header.1", true),
+            new RestHeaderDefinition("header.2", false)));
+        final RestController restController = new RestController(headers, null, null, circuitBreakerService, usageService);
+        Map<String, List<String>> restHeaders = new HashMap<>();
+        restHeaders.put("header.1", Collections.singletonList("boo"));
+        restHeaders.put("header.2", List.of("foo", "bar"));
+        RestRequest fakeRequest = new FakeRestRequest.Builder(xContentRegistry()).withHeaders(restHeaders).build();
+        AssertingChannel channel = new AssertingChannel(fakeRequest, false, RestStatus.BAD_REQUEST);
+        restController.dispatchRequest(fakeRequest, channel, threadContext);
+        assertTrue(channel.getSendResponseCalled());
+    }
+
+    public void testRequestWithDisallowedMultiValuedHeaderButSameValues() {
+        final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        Set<RestHeaderDefinition> headers = new HashSet<>(Arrays.asList(new RestHeaderDefinition("header.1", true),
+            new RestHeaderDefinition("header.2", false)));
+        final RestController restController = new RestController(headers, null, null, circuitBreakerService, usageService);
+        Map<String, List<String>> restHeaders = new HashMap<>();
+        restHeaders.put("header.1", Collections.singletonList("boo"));
+        restHeaders.put("header.2", List.of("foo", "foo"));
+        RestRequest fakeRequest = new FakeRestRequest.Builder(xContentRegistry()).withHeaders(restHeaders).withPath("/bar").build();
+        restController.registerHandler(RestRequest.Method.GET, "/bar", new RestHandler() {
+            @Override
+            public void handleRequest(RestRequest request, RestChannel channel, NodeClient client) throws Exception {
+                channel.sendResponse(new BytesRestResponse(RestStatus.OK, BytesRestResponse.TEXT_CONTENT_TYPE, BytesArray.EMPTY));
+            }
+        });
+        AssertingChannel channel = new AssertingChannel(fakeRequest, false, RestStatus.OK);
+        restController.dispatchRequest(fakeRequest, channel, threadContext);
+        assertTrue(channel.getSendResponseCalled());
+    }
+
     public void testRegisterAsDeprecatedHandler() {
         RestController controller = mock(RestController.class);
 

server/src/test/java/org/elasticsearch/rest/RestRequestTests.java

@@ -41,6 +41,8 @@
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.mockito.Mockito.mock;
@@ -245,6 +247,21 @@ public void testRequiredContent() {
         assertEquals("unknown content type", e.getMessage());
     }
 
+    public void testHeaderHandling() {
+        RestRequest request = contentRestRequest("", emptyMap(),
+            Map.of(
+                "header1", List.of("value1"),
+                "header2", List.of("value1", "value2", "value3"),
+                "header3", List.of("value3", "value3", "value3")));
+
+        assertThat(request.header("header1"), equalTo("value1"));
+        assertThat(request.getSingleValuedHeader("header3"), equalTo("value3"));
+        assertThat(request.getAllHeaderValues("header2"), contains("value1", "value2", "value3"));
+        Exception e = expectThrows(IllegalStateException.class, () -> request.getSingleValuedHeader("header2"));
+        assertThat(e.getMessage(), containsString("multiple values for single-valued header [header2]"));
+        assertThat(request.getAllHeaderValuesAsString("header2"), equalTo("value1,value2,value3"));
+    }
+
     private static RestRequest contentRestRequest(String content, Map<String, String> params) {
         Map<String, List<String>> headers = new HashMap<>();
         headers.put("Content-Type", Collections.singletonList("application/json"));

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/LocalStateCompositeXPackPlugin.java

@@ -62,6 +62,7 @@
 import org.elasticsearch.repositories.Repository;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestHeaderDefinition;
 import org.elasticsearch.script.ScriptContext;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ExecutorBuilder;
@@ -152,8 +153,8 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
     }
 
     @Override
-    public Collection<String> getRestHeaders() {
-        List<String> headers = new ArrayList<>();
+    public Collection<RestHeaderDefinition> getRestHeaders() {
+        List<RestHeaderDefinition> headers = new ArrayList<>();
         headers.addAll(super.getRestHeaders());
         filterPlugins(ActionPlugin.class).stream().forEach(p -> headers.addAll(p.getRestHeaders()));
         return headers;

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

@@ -57,6 +57,7 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.rest.RestHeaderDefinition;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ExecutorBuilder;
 import org.elasticsearch.threadpool.FixedExecutorBuilder;
@@ -619,14 +620,14 @@ public static List<Setting<?>> getSettings(List<SecurityExtension> securityExten
     }
 
     @Override
-    public Collection<String> getRestHeaders() {
-        Set<String> headers = new HashSet<>();
-        headers.add(UsernamePasswordToken.BASIC_AUTH_HEADER);
+    public Collection<RestHeaderDefinition> getRestHeaders() {
+        Set<RestHeaderDefinition> headers = new HashSet<>();
+        headers.add(new RestHeaderDefinition(UsernamePasswordToken.BASIC_AUTH_HEADER, false));
         if (XPackSettings.AUDIT_ENABLED.get(settings)) {
-            headers.add(AuditTrail.X_FORWARDED_FOR_HEADER);
+            headers.add(new RestHeaderDefinition(AuditTrail.X_FORWARDED_FOR_HEADER, true));
         }
         if (AuthenticationServiceField.RUN_AS_ENABLED.get(settings)) {
-            headers.add(AuthenticationServiceField.RUN_AS_USER_HEADER);
+            headers.add(new RestHeaderDefinition(AuthenticationServiceField.RUN_AS_USER_HEADER, false));
         }
         return headers;
     }

x-pack/qa/security-example-spi-extension/src/main/java/org/elasticsearch/example/SpiExtensionPlugin.java

@@ -9,6 +9,7 @@
 import org.elasticsearch.example.realm.CustomRealm;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.RestHeaderDefinition;
 import org.elasticsearch.xpack.core.security.authc.RealmSettings;
 
 import java.util.ArrayList;
@@ -22,8 +23,10 @@
 public class SpiExtensionPlugin extends Plugin implements ActionPlugin {
 
     @Override
-    public Collection<String> getRestHeaders() {
-        return Arrays.asList(CustomRealm.USER_HEADER, CustomRealm.PW_HEADER);
+    public Collection<RestHeaderDefinition> getRestHeaders() {
+        return Arrays.asList(
+            new RestHeaderDefinition(CustomRealm.USER_HEADER, true),
+            new RestHeaderDefinition(CustomRealm.PW_HEADER, true));
     }
 
     @Override