Copy http headers to ThreadContext strictly (#45945) (#48675)

Previous behavior while copying HTTP headers to the ThreadContext,
would allow multiple HTTP headers with the same name, handling only
the first occurrence and disregarding the rest of the values. This
can be confusing when dealing with multiple Headers as it is not
obvious which value is read and which ones are silently dropped.

According to RFC-7230, a client must not send multiple header fields
with the same field name in a HTTP message, unless the entire field
value for this header is defined as a comma separated list or this
specific header is a well-known exception.

This commits changes the behavior in order to be more compliant to
the aforementioned RFC by requiring the classes that implement
ActionPlugin to declare if a header can be multi-valued or not when
registering this header to be copied over to the ThreadContext in
ActionPlugin#getRestHeaders.
If the header is allowed to be multivalued, then all such headers
are read from the HTTP request and their values get concatenated in
a comma-separated string.
If the header is not allowed to be multivalued, and the HTTP
request contains multiple such Headers with different values, the
request is rejected with a 400 status.

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));
         }
     }
 

qa/smoke-test-http/src/test/java/org/elasticsearch/http/ContextAndHeaderTransportIT.java

@@ -50,6 +50,7 @@
 import org.elasticsearch.indices.TermsLookup;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.RestHeaderDefinition;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -81,7 +82,7 @@
 @ClusterScope(scope = SUITE)
 public class ContextAndHeaderTransportIT extends HttpSmokeTestCase {
     private static final List<RequestAndHeaders> requests =  new CopyOnWriteArrayList<>();
-    private static final String CUSTOM_HEADER = "SomeCustomHeader";
+    private static final RestHeaderDefinition CUSTOM_HEADER = new RestHeaderDefinition("SomeCustomHeader", false);
     private String randomHeaderValue = randomAlphaOfLength(20);
     private String queryIndex = "query-" + randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
     private String lookupIndex = "lookup-" + randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
@@ -223,15 +224,15 @@ public void testThatRelevantHttpHeadersBecomeRequestHeaders() throws IOException
         final String IRRELEVANT_HEADER = "SomeIrrelevantHeader";
         Request request = new Request("GET", "/" + queryIndex + "/_search");
         RequestOptions.Builder options = request.getOptions().toBuilder();
-        options.addHeader(CUSTOM_HEADER, randomHeaderValue);
+        options.addHeader(CUSTOM_HEADER.getName(), randomHeaderValue);
         options.addHeader(IRRELEVANT_HEADER, randomHeaderValue);
         request.setOptions(options);
         Response response = getRestClient().performRequest(request);
         assertThat(response.getStatusLine().getStatusCode(), equalTo(200));
         List<RequestAndHeaders> searchRequests = getRequests(SearchRequest.class);
         assertThat(searchRequests, hasSize(greaterThan(0)));
         for (RequestAndHeaders requestAndHeaders : searchRequests) {
-            assertThat(requestAndHeaders.headers.containsKey(CUSTOM_HEADER), is(true));
+            assertThat(requestAndHeaders.headers.containsKey(CUSTOM_HEADER.getName()), is(true));
             // was not specified, thus is not included
             assertThat(requestAndHeaders.headers.containsKey(IRRELEVANT_HEADER), is(false));
         }
@@ -272,21 +273,22 @@ private void assertGetRequestsContainHeaders(String index) {
     }
 
     private void assertRequestContainsHeader(ActionRequest request, Map<String, String> context) {
-        String msg = String.format(Locale.ROOT, "Expected header %s to be in request %s", CUSTOM_HEADER, request.getClass().getName());
+        String msg = String.format(Locale.ROOT, "Expected header %s to be in request %s", CUSTOM_HEADER.getName(),
+            request.getClass().getName());
         if (request instanceof IndexRequest) {
             IndexRequest indexRequest = (IndexRequest) request;
-            msg = String.format(Locale.ROOT, "Expected header %s to be in index request %s/%s/%s", CUSTOM_HEADER,
+            msg = String.format(Locale.ROOT, "Expected header %s to be in index request %s/%s/%s", CUSTOM_HEADER.getName(),
                 indexRequest.index(), indexRequest.type(), indexRequest.id());
         }
-        assertThat(msg, context.containsKey(CUSTOM_HEADER), is(true));
-        assertThat(context.get(CUSTOM_HEADER).toString(), is(randomHeaderValue));
+        assertThat(msg, context.containsKey(CUSTOM_HEADER.getName()), is(true));
+        assertThat(context.get(CUSTOM_HEADER.getName()).toString(), is(randomHeaderValue));
     }
 
     /**
      * a transport client that adds our random header
      */
     private Client transportClient() {
-        return internalCluster().transportClient().filterWithHeader(Collections.singletonMap(CUSTOM_HEADER, randomHeaderValue));
+        return internalCluster().transportClient().filterWithHeader(Collections.singletonMap(CUSTOM_HEADER.getName(), randomHeaderValue));
     }
 
     public static class ActionLoggingPlugin extends Plugin implements ActionPlugin {
@@ -340,7 +342,7 @@ private RequestAndHeaders(Map<String, String> headers, ActionRequest request) {
     }
 
     public static class CustomHeadersPlugin extends Plugin implements ActionPlugin {
-        public Collection<String> getRestHeaders() {
+        public Collection<RestHeaderDefinition> getRestHeaders() {
             return Collections.singleton(CUSTOM_HEADER);
         }
     }

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;
@@ -389,9 +390,9 @@ public ActionModule(boolean transportClient, Settings settings, IndexNameExpress
         actionFilters = setupActionFilters(actionPlugins);
         autoCreateIndex = transportClient ? null : 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

@@ -38,6 +38,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;
@@ -93,7 +94,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

@@ -52,6 +52,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.rest.BytesRestResponse.TEXT_CONTENT_TYPE;
 import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
@@ -73,10 +74,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;
@@ -257,10 +258,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);
+        for (final RestHeaderDefinition restHeader : headersToCopy) {
+            final String name = restHeader.getName();
+            final List<String> headerValues = request.getAllHeaderValues(name);
+            if (headerValues != null && headerValues.isEmpty() == false) {
+                final List<String> distinctHeaderValues = headerValues.stream().distinct().collect(Collectors.toList());
+                if (restHeader.isMultiValueAllowed() == false && distinctHeaderValues.size() > 1) {
+                    channel.sendResponse(
+                        BytesRestResponse.
+                            createSimpleErrorResponse(channel, BAD_REQUEST, "multiple values for single-valued header [" + name + "]."));
+                    return;
+                } else {
+                    threadContext.putHeader(name, String.join(",", distinctHeaderValues));
+                }
             }
         }
         // error_trace cannot be used when we disable detailed errors

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/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", Arrays.asList("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", Arrays.asList("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);
 

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

@@ -63,6 +63,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;
@@ -163,8 +164,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

@@ -59,6 +59,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;
@@ -670,17 +671,17 @@ public static List<Setting<?>> getSettings(boolean transportClientMode, List<Sec
     }
 
     @Override
-    public Collection<String> getRestHeaders() {
+    public Collection<RestHeaderDefinition> getRestHeaders() {
         if (transportClientMode) {
             return Collections.emptyList();
         }
-        Set<String> headers = new HashSet<>();
-        headers.add(UsernamePasswordToken.BASIC_AUTH_HEADER);
+        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, false),
+            new RestHeaderDefinition(CustomRealm.PW_HEADER, false));
     }
 
     @Override