From 78b65e31e84689d433c20852c37ae2d2623add6e Mon Sep 17 00:00:00 2001 From: Brian Clozel Date: Mon, 15 Jan 2018 14:50:58 +0200 Subject: [PATCH] Add support for Forwarded / X-Forwarded-* headers This commit adds new information to the `HttpServerRequest`: * the host (server) address * the remote (client) address * the scheme used by the current request This information can be either derived from the current channel, or extracted from the incoming HTTP request headers using "Forwarded" or "X-Forwarded-*". This feature is opt-in, and must be configured during server setup: HttpServer.create().forwarded().port(8080); Closes gh-220 --- .../ipc/netty/http/server/ConnectionInfo.java | 161 ++++++++++++ .../ipc/netty/http/server/HttpServer.java | 46 +++- .../http/server/HttpServerOperations.java | 36 ++- .../netty/http/server/HttpServerRequest.java | 22 ++ .../http/server/ConnectionInfoTests.java | 238 ++++++++++++++++++ 5 files changed, 497 insertions(+), 6 deletions(-) create mode 100644 src/main/java/reactor/ipc/netty/http/server/ConnectionInfo.java create mode 100644 src/test/java/reactor/ipc/netty/http/server/ConnectionInfoTests.java diff --git a/src/main/java/reactor/ipc/netty/http/server/ConnectionInfo.java b/src/main/java/reactor/ipc/netty/http/server/ConnectionInfo.java new file mode 100644 index 0000000000..f7e8c87c90 --- /dev/null +++ b/src/main/java/reactor/ipc/netty/http/server/ConnectionInfo.java @@ -0,0 +1,161 @@ +/* + * Copyright (c) 2011-2018 Pivotal Software Inc, All Rights Reserved. + * + * Licensed 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 reactor.ipc.netty.http.server; + +import java.net.InetSocketAddress; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.ssl.SslHandler; +import reactor.ipc.netty.tcp.InetSocketAddressUtil; + +/** + * Resolve information about the current connection, including the + * host (server) address, the remote (client) address and the scheme. + * + *

Depending on the chosen factory method, the information + * can be retrieved directly from the channel or additionally + * using the {@code "Forwarded"}, or {@code "X-Forwarded-*"} + * HTTP request headers. + * + * @author Brian Clozel + * @since 0.8 + * @see rfc7239 + */ +public final class ConnectionInfo { + + private static final String FORWARDED_HEADER = "Forwarded"; + private static final Pattern FORWARDED_HOST_PATTERN = Pattern.compile("host=\"?([^;,\"]+)\"?"); + private static final Pattern FORWARDED_PROTO_PATTERN = Pattern.compile("proto=\"?([^;,\"]+)\"?"); + private static final Pattern FORWARDED_FOR_PATTERN = Pattern.compile("for=\"?([^;,\"]+)\"?"); + + private static final String XFORWARDED_IP_HEADER = "X-Forwarded-For"; + private static final String XFORWARDED_HOST_HEADER = "X-Forwarded-Host"; + private static final String XFORWARDED_PORT_HEADER = "X-Forwarded-Port"; + private static final String XFORWARDED_PROTO_HEADER = "X-Forwarded-Proto"; + + private final InetSocketAddress hostAddress; + + private final InetSocketAddress remoteAddress; + + private final String scheme; + + /** + * Retrieve the connection information from the current connection directly + * @param request the current server request + * @param channel the current channel + * @return the connection information + */ + public static ConnectionInfo newConnectionInfo(HttpServerRequest request, SocketChannel channel) { + InetSocketAddress hostAddress = channel.localAddress(); + InetSocketAddress remoteAddress = channel.remoteAddress(); + String scheme = channel.pipeline().get(SslHandler.class) != null ? "https" : "http"; + return new ConnectionInfo(hostAddress, remoteAddress, scheme); + } + + /** + * Retrieve the connection information from the {@code "Forwarded"}/{@code "X-Forwarded-*"} + * HTTP request headers, or from the current connection directly if none are found. + * @param request the current server request + * @param channel the current channel + * @return the connection information + */ + public static ConnectionInfo newForwardedConnectionInfo(HttpServerRequest request, SocketChannel channel) { + if (request.requestHeaders().contains(FORWARDED_HEADER)) { + return parseForwardedInfo(request, channel); + } + else { + return parseXForwardedInfo(request, channel); + } + } + + private static ConnectionInfo parseForwardedInfo(HttpServerRequest request, SocketChannel channel) { + InetSocketAddress hostAddress = channel.localAddress(); + InetSocketAddress remoteAddress = channel.remoteAddress(); + String scheme = channel.pipeline().get(SslHandler.class) != null ? "https" : "http"; + + String forwarded = request.requestHeaders().get(FORWARDED_HEADER).split(",")[0]; + Matcher hostMatcher = FORWARDED_HOST_PATTERN.matcher(forwarded); + if (hostMatcher.find()) { + hostAddress = parseAddress(hostMatcher.group(1), hostAddress.getPort()); + } + Matcher protoMatcher = FORWARDED_PROTO_PATTERN.matcher(forwarded); + if (protoMatcher.find()) { + scheme = protoMatcher.group(1).trim(); + } + Matcher forMatcher = FORWARDED_FOR_PATTERN.matcher(forwarded); + if(forMatcher.find()) { + remoteAddress = parseAddress(forMatcher.group(1).trim(), remoteAddress.getPort()); + } + return new ConnectionInfo(hostAddress, remoteAddress, scheme); + } + + private static InetSocketAddress parseAddress(String address, int defaultPort) { + int portSeparatorIdx = address.lastIndexOf(":"); + if (portSeparatorIdx > address.lastIndexOf("]")) { + return InetSocketAddressUtil.createUnresolved(address.substring(0, portSeparatorIdx), + Integer.parseInt(address.substring(portSeparatorIdx + 1))); + } + else { + return InetSocketAddressUtil.createUnresolved(address, defaultPort); + } + } + + private static ConnectionInfo parseXForwardedInfo(HttpServerRequest request, SocketChannel channel) { + InetSocketAddress hostAddress = channel.localAddress(); + InetSocketAddress remoteAddress = channel.remoteAddress(); + String scheme = channel.pipeline().get(SslHandler.class) != null ? "https" : "http"; + if (request.requestHeaders().contains(XFORWARDED_IP_HEADER)) { + String hostValue = request.requestHeaders().get(XFORWARDED_IP_HEADER).split(",")[0]; + hostAddress = parseAddress(hostValue, hostAddress.getPort()); + } + else if(request.requestHeaders().contains(XFORWARDED_HOST_HEADER)) { + if(request.requestHeaders().contains(XFORWARDED_PORT_HEADER)) { + hostAddress = InetSocketAddressUtil.createUnresolved( + request.requestHeaders().get(XFORWARDED_HOST_HEADER).split(",")[0].trim(), + Integer.parseInt(request.requestHeaders().get(XFORWARDED_PORT_HEADER).split(",")[0].trim())); + } + else { + hostAddress = InetSocketAddressUtil.createUnresolved( + request.requestHeaders().get(XFORWARDED_HOST_HEADER).split(",")[0].trim(), + channel.localAddress().getPort()); + } + } + if (request.requestHeaders().contains(XFORWARDED_PROTO_HEADER)) { + scheme = request.requestHeaders().get(XFORWARDED_PROTO_HEADER).trim(); + } + return new ConnectionInfo(hostAddress, remoteAddress, scheme); + } + + private ConnectionInfo(InetSocketAddress hostAddress, InetSocketAddress remoteAddress, String scheme) { + this.hostAddress = hostAddress; + this.remoteAddress = remoteAddress; + this.scheme = scheme; + } + + public InetSocketAddress getHostAddress() { + return hostAddress; + } + + public InetSocketAddress getRemoteAddress() { + return remoteAddress; + } + + public String getScheme() { + return scheme; + } +} diff --git a/src/main/java/reactor/ipc/netty/http/server/HttpServer.java b/src/main/java/reactor/ipc/netty/http/server/HttpServer.java index df671371b4..b887440ae5 100644 --- a/src/main/java/reactor/ipc/netty/http/server/HttpServer.java +++ b/src/main/java/reactor/ipc/netty/http/server/HttpServer.java @@ -177,6 +177,16 @@ public final HttpServer compress() { return tcpConfiguration(COMPRESS_ATTR_CONFIG); } + /** + * Enable support for the {@code "Forwarded"} and {@code "X-Forwarded-*"} + * HTTP request headers for deriving information about the connection. + * + * @return a new {@link HttpServer} + */ + public final HttpServer forwarded() { + return tcpConfiguration(FORWARDED_ATTR_CONFIG); + } + /** * Enable GZip response compression if the client request presents accept encoding * headers @@ -225,6 +235,16 @@ public final HttpServer noCompression() { return tcpConfiguration(COMPRESS_ATTR_DISABLE); } + /** + * Disable support for the {@code "Forwarded"} and {@code "X-Forwarded-*"} + * HTTP request headers. + * + * @return a new {@link HttpServer} + */ + public final HttpServer noForwarded() { + return tcpConfiguration(FORWARDED_ATTR_DISABLE); + } + /** * Apply {@link ServerBootstrap} configuration given mapper taking currently * configured one and returning a new one to be ultimately used for socket binding. @@ -277,7 +297,20 @@ protected TcpServer tcpConfiguration() { @Nullable @Override public ChannelOperations create(Connection c, ConnectionEvents listener, Object msg) { - return HttpServerOperations.bindHttp(c, listener, msg); + return HttpServerOperations.bindHttp(c, listener, msg, false); + } + + @Override + public boolean createOnConnected() { + return false; + } + }; + + static final ChannelOperations.OnSetup HTTP_OPS_FORWARDED = new ChannelOperations.OnSetup() { + @Nullable + @Override + public ChannelOperations create(Connection c, ConnectionEvents listener, Object msg) { + return HttpServerOperations.bindHttp(c, listener, msg, true); } @Override @@ -294,6 +327,11 @@ public boolean createOnConnected() { return b; }; + static final Function HTTP_OPS_FORWARDED_CONF = b -> { + BootstrapHandlers.channelOperationFactory(b, HTTP_OPS_FORWARDED); + return b; + }; + static final TcpServer DEFAULT_TCP_SERVER = TcpServer.create() .bootstrap(HTTP_OPS_CONF) .port(DEFAULT_PORT); @@ -306,4 +344,10 @@ public boolean createOnConnected() { static final Function COMPRESS_ATTR_DISABLE = tcp -> tcp.selectorAttr(HttpServerBind.PRODUCE_GZIP, null); + + static final Function FORWARDED_ATTR_CONFIG = + tcp -> tcp.bootstrap(HTTP_OPS_FORWARDED_CONF); + + static final Function FORWARDED_ATTR_DISABLE = + tcp -> tcp.bootstrap(HTTP_OPS_CONF); } diff --git a/src/main/java/reactor/ipc/netty/http/server/HttpServerOperations.java b/src/main/java/reactor/ipc/netty/http/server/HttpServerOperations.java index 08272ff299..6adb3f3898 100644 --- a/src/main/java/reactor/ipc/netty/http/server/HttpServerOperations.java +++ b/src/main/java/reactor/ipc/netty/http/server/HttpServerOperations.java @@ -17,6 +17,7 @@ package reactor.ipc.netty.http.server; import java.io.IOException; +import java.net.InetSocketAddress; import java.nio.file.Files; import java.nio.file.Path; import java.util.Map; @@ -25,11 +26,13 @@ import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; + import javax.annotation.Nullable; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.DefaultHttpResponse; import io.netty.handler.codec.http.FullHttpResponse; @@ -48,6 +51,7 @@ import io.netty.handler.codec.http.cookie.Cookie; import io.netty.handler.codec.http.cookie.ServerCookieEncoder; import io.netty.util.AsciiString; +import io.netty.util.AttributeKey; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -74,9 +78,9 @@ class HttpServerOperations extends HttpOperations> paramsResolver; HttpServerOperations(HttpServerOperations replaced) { super(replaced); this.cookieHolder = replaced.cookieHolder; + this.connectionInfo = replaced.connectionInfo; this.responseHeaders = replaced.responseHeaders; this.nettyResponse = replaced.nettyResponse; this.paramsResolver = replaced.paramsResolver; @@ -101,15 +107,20 @@ static HttpServerOperations bindHttp(Connection connection, ConnectionEvents lis HttpServerOperations(Connection c, ConnectionEvents listener, - HttpRequest nettyRequest) { + HttpRequest nettyRequest, + boolean forwarded) { super(c, listener); this.nettyRequest = Objects.requireNonNull(nettyRequest, "nettyRequest"); this.nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); this.responseHeaders = nettyResponse.headers(); this.cookieHolder = Cookies.newServerRequestHolder(requestHeaders()); + if (forwarded) { + this.connectionInfo = ConnectionInfo.newForwardedConnectionInfo(this, (SocketChannel) channel()); + } + else { + this.connectionInfo = ConnectionInfo.newConnectionInfo(this, (SocketChannel) channel()); + } chunkedTransfer(true); - - } @Override @@ -259,6 +270,16 @@ public Flux receiveObject() { } } + @Override + public InetSocketAddress hostAddress() { + return this.connectionInfo.getHostAddress(); + } + + @Override + public InetSocketAddress remoteAddress() { + return this.connectionInfo.getRemoteAddress(); + } + @Override public HttpHeaders requestHeaders() { if (nettyRequest != null) { @@ -267,6 +288,11 @@ public HttpHeaders requestHeaders() { throw new IllegalStateException("request not parsed"); } + @Override + public String scheme() { + return this.connectionInfo.getScheme(); + } + @Override public HttpHeaders responseHeaders() { return responseHeaders; diff --git a/src/main/java/reactor/ipc/netty/http/server/HttpServerRequest.java b/src/main/java/reactor/ipc/netty/http/server/HttpServerRequest.java index 49e54929e8..12444995e5 100644 --- a/src/main/java/reactor/ipc/netty/http/server/HttpServerRequest.java +++ b/src/main/java/reactor/ipc/netty/http/server/HttpServerRequest.java @@ -16,6 +16,7 @@ package reactor.ipc.netty.http.server; +import java.net.InetSocketAddress; import java.util.Map; import java.util.function.Consumer; import java.util.function.Function; @@ -77,6 +78,20 @@ default Flux receiveContent() { return receiveObject().ofType(HttpContent.class); } + /** + * Return the address of the host peer. + * + * @return the host's address + */ + InetSocketAddress hostAddress(); + + /** + * Return the address of the remote peer. + * + * @return the peer's address + */ + InetSocketAddress remoteAddress(); + /** * Return inbound {@link HttpHeaders} * @@ -84,4 +99,11 @@ default Flux receiveContent() { */ HttpHeaders requestHeaders(); + /** + * Return the current scheme + * + * @return the protocol scheme + */ + String scheme(); + } diff --git a/src/test/java/reactor/ipc/netty/http/server/ConnectionInfoTests.java b/src/test/java/reactor/ipc/netty/http/server/ConnectionInfoTests.java new file mode 100644 index 0000000000..1ddc737c13 --- /dev/null +++ b/src/test/java/reactor/ipc/netty/http/server/ConnectionInfoTests.java @@ -0,0 +1,238 @@ +/* + * Copyright (c) 2011-2018 Pivotal Software Inc, All Rights Reserved. + * + * Licensed 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 reactor.ipc.netty.http.server; + + +import java.util.function.Consumer; + +import io.netty.handler.codec.http.HttpMethod; +import org.junit.After; +import org.junit.Test; +import reactor.core.publisher.Mono; +import reactor.ipc.netty.DisposableServer; +import reactor.ipc.netty.http.client.HttpClient; +import reactor.ipc.netty.http.client.HttpClientRequest; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link ConnectionInfo} + * + * @author Brian Clozel + */ +public class ConnectionInfoTests { + + private DisposableServer connection; + + @Test + public void noHeaders() { + testClientRequest( + clientRequest -> { + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isIn("0:0:0:0:0:0:0:1", "127.0.0.1"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(this.connection.address().getPort()); + }); + } + + @Test + public void forwardedHost() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "host=192.168.0.1"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("192.168.0.1"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(this.connection.address().getPort()); + }); + } + + @Test + public void forwardedHostIpV6() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "host=[1abc:2abc:3abc::5ABC:6abc]"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("1abc:2abc:3abc:0:0:0:5abc:6abc"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(this.connection.address().getPort()); + }); + } + + @Test + public void xForwardedFor() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("X-Forwarded-For", "[1abc:2abc:3abc::5ABC:6abc]:8080, 192.168.0.1"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("1abc:2abc:3abc:0:0:0:5abc:6abc"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(8080); + }); + } + + @Test + public void xForwardedHost() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("X-Forwarded-Host", "[1abc:2abc:3abc::5ABC:6abc], 192.168.0.1"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("1abc:2abc:3abc:0:0:0:5abc:6abc"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(this.connection.address().getPort()); + }); + } + + @Test + public void xForwardedHostAndPort() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("X-Forwarded-Host", "192.168.0.1"); + clientRequest.addHeader("X-Forwarded-Port", "8080"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("192.168.0.1"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(8080); + }); + } + + @Test + public void forwardedMultipleHosts() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "host=a.example.com,host=b.example.com, host=c.example.com"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("a.example.com"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(this.connection.address().getPort()); + }); + } + + @Test + public void forwardedAllDirectives() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "host=a.example.com:443;proto=https"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("a.example.com"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(443); + assertThat(serverRequest.scheme()).isEqualTo("https"); + }); + } + + @Test + public void forwardedAllDirectivesQuoted() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "host=\"a.example.com:443\";proto=\"https\""); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("a.example.com"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(443); + assertThat(serverRequest.scheme()).isEqualTo("https"); + }); + } + + @Test + public void forwardedMultipleHeaders() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "host=a.example.com:443;proto=https"); + clientRequest.addHeader("Forwarded", "host=b.example.com"); + }, + serverRequest -> { + assertThat(serverRequest.hostAddress().getHostString()).isEqualTo("a.example.com"); + assertThat(serverRequest.hostAddress().getPort()).isEqualTo(443); + assertThat(serverRequest.scheme()).isEqualTo("https"); + }); + } + + @Test + public void forwardedForHostname() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "for=\"_gazonk\""); + }, + serverRequest -> { + assertThat(serverRequest.remoteAddress().getHostString()).isEqualTo("_gazonk"); + assertThat(serverRequest.remoteAddress().getPort()).isPositive(); + }); + } + + @Test + public void forwardedForIp() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "for=192.0.2.60;proto=http;by=203.0.113.43"); + }, + serverRequest -> { + assertThat(serverRequest.remoteAddress().getHostString()).isEqualTo("192.0.2.60"); + assertThat(serverRequest.remoteAddress().getPort()).isPositive(); + assertThat(serverRequest.scheme()).isEqualTo("http"); + }); + } + + @Test + public void forwardedForIpV6() { + testClientRequest( + clientRequest -> { + clientRequest.addHeader("Forwarded", "for=\"[2001:db8:cafe::17]:4711\""); + }, + serverRequest -> { + assertThat(serverRequest.remoteAddress().getHostString()).isEqualTo("2001:db8:cafe:0:0:0:0:17"); + assertThat(serverRequest.remoteAddress().getPort()).isEqualTo(4711); + }); + } + + + private void testClientRequest(Consumer clientConsumer, + Consumer serverConsumer) { + + this.connection = HttpServer.create().forwarded().port(0).handler((req, res) -> { + try { + serverConsumer.accept(req); + return res.status(200).sendString(Mono.just("OK")); + } + catch (Throwable e) { + return res.status(500).sendString(Mono.just(e.getMessage())); + } + }).bindNow(); + + String response = + HttpClient.prepare() + .port(this.connection.address().getPort()) + .wiretap() + .request(HttpMethod.GET) + .uri("/test") + .send((req, out) -> { + clientConsumer.accept(req); + return req.sendHeaders(); + }) + .responseContent() + .aggregate() + .asString() + .block(); + + assertThat(response).isEqualTo("OK"); + } + + @After + public void tearDown() { + this.connection.dispose(); + } + +} \ No newline at end of file