Add cors support to NioHttpServerTransport (elastic#30827)

This is related to elastic#28898. This commit adds cors support to the nio http
transport. Most of the work is copied directly from the netty module
implementation. Additionally, this commit adds tests for the nio http
channel.

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/cors/Netty4CorsConfig.java

@@ -76,7 +76,8 @@ public boolean isCorsSupportEnabled() {
     }
 
     /**
-     * Determines whether a wildcard origin, '*', is supported.
+     * Determines whether a wildcard origin, '*', is supported. This also means that null origins are
+     * supported.
      *
      * @return {@code boolean} true if any origin is allowed.
      */
@@ -121,21 +122,21 @@ public boolean isNullOriginAllowed() {
     }
 
     /**
-     * Determines if cookies are supported for CORS requests.
+     * Determines if credentials are supported for CORS requests.
      *
-     * By default cookies are not included in CORS requests but if isCredentialsAllowed returns
-     * true cookies will be added to CORS requests. Setting this value to true will set the
+     * By default credentials are not included in CORS requests but if isCredentialsAllowed returns
+     * true credentials will be added to CORS requests. Setting this value to true will set the
      * CORS 'Access-Control-Allow-Credentials' response header to true.
      *
-     * Please note that cookie support needs to be enabled on the client side as well.
-     * The client needs to opt-in to send cookies by calling:
+     * Please note that credentials support needs to be enabled on the client side as well.
+     * The client needs to opt-in to send credentials by calling:
      * <pre>
      * xhr.withCredentials = true;
      * </pre>
-     * The default value for 'withCredentials' is false in which case no cookies are sent.
-     * Setting this to true will included cookies in cross origin requests.
+     * The default value for 'withCredentials' is false in which case no credentials are sent.
+     * Setting this to true will included credentials in cross origin requests.
      *
-     * @return {@code true} if cookies are supported.
+     * @return {@code true} if credentials are supported.
      */
     public boolean isCredentialsAllowed() {
         return allowCredentials;

plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/HttpReadWriteHandler.java

@@ -36,6 +36,8 @@
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.http.HttpHandlingSettings;
 import org.elasticsearch.http.HttpPipelinedRequest;
+import org.elasticsearch.http.nio.cors.NioCorsConfig;
+import org.elasticsearch.http.nio.cors.NioCorsHandler;
 import org.elasticsearch.nio.FlushOperation;
 import org.elasticsearch.nio.InboundChannelBuffer;
 import org.elasticsearch.nio.NioSocketChannel;
@@ -50,21 +52,25 @@
 import java.util.List;
 import java.util.function.BiConsumer;
 
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ENABLED;
+
 public class HttpReadWriteHandler implements ReadWriteHandler {
 
     private final NettyAdaptor adaptor;
     private final NioSocketChannel nioChannel;
     private final NioHttpServerTransport transport;
     private final HttpHandlingSettings settings;
     private final NamedXContentRegistry xContentRegistry;
+    private final NioCorsConfig corsConfig;
     private final ThreadContext threadContext;
 
     HttpReadWriteHandler(NioSocketChannel nioChannel, NioHttpServerTransport transport, HttpHandlingSettings settings,
-                         NamedXContentRegistry xContentRegistry, ThreadContext threadContext) {
+                         NamedXContentRegistry xContentRegistry, NioCorsConfig corsConfig, ThreadContext threadContext) {
         this.nioChannel = nioChannel;
         this.transport = transport;
         this.settings = settings;
         this.xContentRegistry = xContentRegistry;
+        this.corsConfig = corsConfig;
         this.threadContext = threadContext;
 
         List<ChannelHandler> handlers = new ArrayList<>(5);
@@ -78,6 +84,9 @@ public class HttpReadWriteHandler implements ReadWriteHandler {
         if (settings.isCompression()) {
             handlers.add(new HttpContentCompressor(settings.getCompressionLevel()));
         }
+        if (settings.isCorsEnabled()) {
+            handlers.add(new NioCorsHandler(corsConfig));
+        }
         handlers.add(new NioHttpPipeliningHandler(transport.getLogger(), settings.getPipeliningMaxEvents()));
 
         adaptor = new NettyAdaptor(handlers.toArray(new ChannelHandler[0]));
@@ -178,7 +187,7 @@ private void handleRequest(Object msg) {
                 int sequence = pipelinedRequest.getSequence();
                 BigArrays bigArrays = transport.getBigArrays();
                 try {
-                    innerChannel = new NioHttpChannel(nioChannel, bigArrays, httpRequest, sequence, settings, threadContext);
+                    innerChannel = new NioHttpChannel(nioChannel, bigArrays, httpRequest, sequence, settings, corsConfig, threadContext);
                 } catch (final IllegalArgumentException e) {
                     if (badRequestCause == null) {
                         badRequestCause = e;
@@ -191,7 +200,7 @@ private void handleRequest(Object msg) {
                             Collections.emptyMap(), // we are going to dispatch the request as a bad request, drop all parameters
                             copiedRequest.uri(),
                             copiedRequest);
-                    innerChannel = new NioHttpChannel(nioChannel, bigArrays, innerRequest, sequence, settings, threadContext);
+                    innerChannel = new NioHttpChannel(nioChannel, bigArrays, innerRequest, sequence, settings, corsConfig, threadContext);
                 }
                 channel = innerChannel;
             }

plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NioHttpChannel.java

@@ -41,6 +41,8 @@
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.http.HttpHandlingSettings;
+import org.elasticsearch.http.nio.cors.NioCorsConfig;
+import org.elasticsearch.http.nio.cors.NioCorsHandler;
 import org.elasticsearch.nio.NioSocketChannel;
 import org.elasticsearch.rest.AbstractRestChannel;
 import org.elasticsearch.rest.RestResponse;
@@ -58,17 +60,19 @@ public class NioHttpChannel extends AbstractRestChannel {
 
     private final BigArrays bigArrays;
     private final int sequence;
+    private final NioCorsConfig corsConfig;
     private final ThreadContext threadContext;
     private final FullHttpRequest nettyRequest;
     private final NioSocketChannel nioChannel;
     private final boolean resetCookies;
 
     NioHttpChannel(NioSocketChannel nioChannel, BigArrays bigArrays, NioHttpRequest request, int sequence,
-                   HttpHandlingSettings settings, ThreadContext threadContext) {
+                   HttpHandlingSettings settings, NioCorsConfig corsConfig, ThreadContext threadContext) {
         super(request, settings.getDetailedErrorsEnabled());
         this.nioChannel = nioChannel;
         this.bigArrays = bigArrays;
         this.sequence = sequence;
+        this.corsConfig = corsConfig;
         this.threadContext = threadContext;
         this.nettyRequest = request.getRequest();
         this.resetCookies = settings.isResetCookies();
@@ -87,6 +91,8 @@ public void sendResponse(RestResponse response) {
         }
         resp.setStatus(getStatus(response.status()));
 
+        NioCorsHandler.setCorsResponseHeaders(nettyRequest, resp, corsConfig);
+
         String opaque = nettyRequest.headers().get("X-Opaque-Id");
         if (opaque != null) {
             setHeaderField(resp, "X-Opaque-Id", opaque);

plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NioHttpServerTransport.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.http.nio;
 
+import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.timeout.ReadTimeoutException;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
@@ -28,6 +29,7 @@
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Setting;
@@ -38,11 +40,13 @@
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.http.AbstractHttpServerTransport;
 import org.elasticsearch.http.BindHttpException;
 import org.elasticsearch.http.HttpHandlingSettings;
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.http.HttpStats;
-import org.elasticsearch.http.AbstractHttpServerTransport;
+import org.elasticsearch.http.nio.cors.NioCorsConfig;
+import org.elasticsearch.http.nio.cors.NioCorsConfigBuilder;
 import org.elasticsearch.nio.AcceptingSelector;
 import org.elasticsearch.nio.AcceptorEventHandler;
 import org.elasticsearch.nio.BytesChannelContext;
@@ -56,6 +60,7 @@
 import org.elasticsearch.nio.SocketChannelContext;
 import org.elasticsearch.nio.SocketEventHandler;
 import org.elasticsearch.nio.SocketSelector;
+import org.elasticsearch.rest.RestUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
@@ -64,15 +69,23 @@
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
+import java.util.regex.Pattern;
 
 import static org.elasticsearch.common.settings.Setting.intSetting;
 import static org.elasticsearch.common.util.concurrent.EsExecutors.daemonThreadFactory;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_CREDENTIALS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_HEADERS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_METHODS;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_ORIGIN;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ENABLED;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_MAX_AGE;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_COMPRESSION;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_COMPRESSION_LEVEL;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_DETAILED_ERRORS_ENABLED;
@@ -86,6 +99,7 @@
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_REUSE_ADDRESS;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_TCP_SEND_BUFFER_SIZE;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_PIPELINING_MAX_EVENTS;
+import static org.elasticsearch.http.nio.cors.NioCorsHandler.ANY_ORIGIN;
 
 public class NioHttpServerTransport extends AbstractHttpServerTransport {
 
@@ -115,6 +129,7 @@ public class NioHttpServerTransport extends AbstractHttpServerTransport {
     private final Set<NioSocketChannel> socketChannels = Collections.newSetFromMap(new ConcurrentHashMap<>());
     private NioGroup nioGroup;
     private HttpChannelFactory channelFactory;
+    private final NioCorsConfig corsConfig;
 
     public NioHttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays, ThreadPool threadPool,
                                   NamedXContentRegistry xContentRegistry, HttpServerTransport.Dispatcher dispatcher) {
@@ -136,6 +151,7 @@ public NioHttpServerTransport(Settings settings, NetworkService networkService,
             SETTING_HTTP_COMPRESSION_LEVEL.get(settings),
             SETTING_HTTP_DETAILED_ERRORS_ENABLED.get(settings),
             pipeliningMaxEvents);
+        this.corsConfig = buildCorsConfig(settings);
 
         this.tcpNoDelay = SETTING_HTTP_TCP_NO_DELAY.get(settings);
         this.tcpKeepAlive = SETTING_HTTP_TCP_KEEP_ALIVE.get(settings);
@@ -279,6 +295,38 @@ protected void nonChannelExceptionCaught(Exception ex) {
         logger.warn(new ParameterizedMessage("exception caught on transport layer [thread={}]", Thread.currentThread().getName()), ex);
     }
 
+    static NioCorsConfig buildCorsConfig(Settings settings) {
+        if (SETTING_CORS_ENABLED.get(settings) == false) {
+            return NioCorsConfigBuilder.forOrigins().disable().build();
+        }
+        String origin = SETTING_CORS_ALLOW_ORIGIN.get(settings);
+        final NioCorsConfigBuilder builder;
+        if (Strings.isNullOrEmpty(origin)) {
+            builder = NioCorsConfigBuilder.forOrigins();
+        } else if (origin.equals(ANY_ORIGIN)) {
+            builder = NioCorsConfigBuilder.forAnyOrigin();
+        } else {
+            Pattern p = RestUtils.checkCorsSettingForRegex(origin);
+            if (p == null) {
+                builder = NioCorsConfigBuilder.forOrigins(RestUtils.corsSettingAsArray(origin));
+            } else {
+                builder = NioCorsConfigBuilder.forPattern(p);
+            }
+        }
+        if (SETTING_CORS_ALLOW_CREDENTIALS.get(settings)) {
+            builder.allowCredentials();
+        }
+        String[] strMethods = Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_METHODS.get(settings), ",");
+        HttpMethod[] methods = Arrays.stream(strMethods)
+            .map(HttpMethod::valueOf)
+            .toArray(HttpMethod[]::new);
+        return builder.allowedRequestMethods(methods)
+            .maxAge(SETTING_CORS_MAX_AGE.get(settings))
+            .allowedRequestHeaders(Strings.tokenizeToStringArray(SETTING_CORS_ALLOW_HEADERS.get(settings), ","))
+            .shortCircuit()
+            .build();
+    }
+
     private void closeChannels(List<NioChannel> channels) {
         List<ActionFuture<Void>> futures = new ArrayList<>(channels.size());
 
@@ -315,7 +363,7 @@ private HttpChannelFactory() {
         public NioSocketChannel createChannel(SocketSelector selector, SocketChannel channel) throws IOException {
             NioSocketChannel nioChannel = new NioSocketChannel(channel);
             HttpReadWriteHandler httpReadWritePipeline = new HttpReadWriteHandler(nioChannel,NioHttpServerTransport.this,
-                httpHandlingSettings, xContentRegistry, threadPool.getThreadContext());
+                httpHandlingSettings, xContentRegistry, corsConfig, threadPool.getThreadContext());
             Consumer<Exception> exceptionHandler = (e) -> exceptionCaught(nioChannel, e);
             SocketChannelContext context = new BytesChannelContext(nioChannel, selector, exceptionHandler, httpReadWritePipeline,
                 InboundChannelBuffer.allocatingInstance());