Skip to content

Commit 305bfea

Browse files
Tim-BrooksTim-Brooks
authored
Add nio http transport to security plugin (elastic#32018)
This is related to elastic#27260. It adds the SecurityNioHttpServerTransport to the security plugin. It randomly uses the nio http transport in security integration tests.
1 parent 3679d00 commit 305bfea

23 files changed

+750
-156
lines changed

‎modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ public void sendMessage(BytesReference reference, ActionListener<Void> listener)
112112
}
113113
}
114114

115-
public Channel getLowLevelChannel() {
115+
public Channel getNettyChannel() {
116116
return channel;
117117
}
118118

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,8 @@ public class HttpReadWriteHandler implements ReadWriteHandler {
5151
private final NioHttpChannel nioHttpChannel;
5252
private final NioHttpServerTransport transport;
5353

54-
HttpReadWriteHandler(NioHttpChannel nioHttpChannel, NioHttpServerTransport transport, HttpHandlingSettings settings,
55-
NioCorsConfig corsConfig) {
54+
public HttpReadWriteHandler(NioHttpChannel nioHttpChannel, NioHttpServerTransport transport, HttpHandlingSettings settings,
55+
NioCorsConfig corsConfig) {
5656
this.nioHttpChannel = nioHttpChannel;
5757
this.transport = transport;
5858

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828

2929
public class NioHttpChannel extends NioSocketChannel implements HttpChannel {
3030

31-
NioHttpChannel(SocketChannel socketChannel) {
31+
public NioHttpChannel(SocketChannel socketChannel) {
3232
super(socketChannel);
3333
}
3434

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,12 +23,11 @@
2323
import org.elasticsearch.http.HttpServerChannel;
2424
import org.elasticsearch.nio.NioServerSocketChannel;
2525

26-
import java.io.IOException;
2726
import java.nio.channels.ServerSocketChannel;
2827

2928
public class NioHttpServerChannel extends NioServerSocketChannel implements HttpServerChannel {
3029

31-
NioHttpServerChannel(ServerSocketChannel serverSocketChannel) throws IOException {
30+
public NioHttpServerChannel(ServerSocketChannel serverSocketChannel) {
3231
super(serverSocketChannel);
3332
}
3433

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

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@
3535
import org.elasticsearch.http.AbstractHttpServerTransport;
3636
import org.elasticsearch.http.HttpChannel;
3737
import org.elasticsearch.http.HttpServerChannel;
38-
import org.elasticsearch.http.HttpServerTransport;
3938
import org.elasticsearch.http.nio.cors.NioCorsConfig;
4039
import org.elasticsearch.http.nio.cors.NioCorsConfigBuilder;
4140
import org.elasticsearch.nio.BytesChannelContext;
@@ -87,21 +86,21 @@ public class NioHttpServerTransport extends AbstractHttpServerTransport {
8786
(s) -> Integer.toString(EsExecutors.numberOfProcessors(s) * 2),
8887
(s) -> Setting.parseInt(s, 1, "http.nio.worker_count"), Setting.Property.NodeScope);
8988

90-
private final PageCacheRecycler pageCacheRecycler;
89+
protected final PageCacheRecycler pageCacheRecycler;
90+
protected final NioCorsConfig corsConfig;
9191

92-
private final boolean tcpNoDelay;
93-
private final boolean tcpKeepAlive;
94-
private final boolean reuseAddress;
95-
private final int tcpSendBufferSize;
96-
private final int tcpReceiveBufferSize;
92+
protected final boolean tcpNoDelay;
93+
protected final boolean tcpKeepAlive;
94+
protected final boolean reuseAddress;
95+
protected final int tcpSendBufferSize;
96+
protected final int tcpReceiveBufferSize;
9797

9898
private NioGroup nioGroup;
99-
private HttpChannelFactory channelFactory;
100-
private final NioCorsConfig corsConfig;
99+
private ChannelFactory<NioHttpServerChannel, NioHttpChannel> channelFactory;
101100

102101
public NioHttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays,
103102
PageCacheRecycler pageCacheRecycler, ThreadPool threadPool, NamedXContentRegistry xContentRegistry,
104-
HttpServerTransport.Dispatcher dispatcher) {
103+
Dispatcher dispatcher) {
105104
super(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher);
106105
this.pageCacheRecycler = pageCacheRecycler;
107106

@@ -136,7 +135,7 @@ protected void doStart() {
136135
nioGroup = new NioGroup(daemonThreadFactory(this.settings, HTTP_SERVER_ACCEPTOR_THREAD_NAME_PREFIX), acceptorCount,
137136
daemonThreadFactory(this.settings, HTTP_SERVER_WORKER_THREAD_NAME_PREFIX), workerCount,
138137
(s) -> new EventHandler(this::onNonChannelException, s));
139-
channelFactory = new HttpChannelFactory();
138+
channelFactory = channelFactory();
140139
bindServer();
141140
success = true;
142141
} catch (IOException e) {
@@ -162,6 +161,10 @@ protected HttpServerChannel bind(InetSocketAddress socketAddress) throws IOExcep
162161
return nioGroup.bindServerChannel(socketAddress, channelFactory);
163162
}
164163

164+
protected ChannelFactory<NioHttpServerChannel, NioHttpChannel> channelFactory() {
165+
return new HttpChannelFactory();
166+
}
167+
165168
static NioCorsConfig buildCorsConfig(Settings settings) {
166169
if (SETTING_CORS_ENABLED.get(settings) == false) {
167170
return NioCorsConfigBuilder.forOrigins().disable().build();
@@ -194,7 +197,7 @@ static NioCorsConfig buildCorsConfig(Settings settings) {
194197
.build();
195198
}
196199

197-
private void acceptChannel(NioSocketChannel socketChannel) {
200+
protected void acceptChannel(NioSocketChannel socketChannel) {
198201
super.serverAcceptedChannel((HttpChannel) socketChannel);
199202
}
200203

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

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -200,11 +200,13 @@
200200
import org.elasticsearch.xpack.security.rest.action.user.RestPutUserAction;
201201
import org.elasticsearch.xpack.security.rest.action.user.RestSetEnabledAction;
202202
import org.elasticsearch.xpack.security.support.SecurityIndexManager;
203+
import org.elasticsearch.xpack.security.transport.SecurityHttpSettings;
203204
import org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor;
204205
import org.elasticsearch.xpack.security.transport.filter.IPFilter;
205206
import org.elasticsearch.xpack.security.transport.netty4.SecurityNetty4HttpServerTransport;
206207
import org.elasticsearch.xpack.security.transport.netty4.SecurityNetty4ServerTransport;
207208
import org.elasticsearch.xpack.core.template.TemplateUtils;
209+
import org.elasticsearch.xpack.security.transport.nio.SecurityNioHttpServerTransport;
208210
import org.elasticsearch.xpack.security.transport.nio.SecurityNioTransport;
209211
import org.joda.time.DateTime;
210212
import org.joda.time.DateTimeZone;
@@ -511,21 +513,22 @@ static Settings additionalSettings(final Settings settings, final boolean enable
511513

512514
if (NetworkModule.HTTP_TYPE_SETTING.exists(settings)) {
513515
final String httpType = NetworkModule.HTTP_TYPE_SETTING.get(settings);
514-
if (httpType.equals(SecurityField.NAME4)) {
515-
SecurityNetty4HttpServerTransport.overrideSettings(builder, settings);
516+
if (httpType.equals(SecurityField.NAME4) || httpType.equals(SecurityField.NIO)) {
517+
SecurityHttpSettings.overrideSettings(builder, settings);
516518
} else {
517519
final String message = String.format(
518520
Locale.ROOT,
519-
"http type setting [%s] must be [%s] but is [%s]",
521+
"http type setting [%s] must be [%s] or [%s] but is [%s]",
520522
NetworkModule.HTTP_TYPE_KEY,
521523
SecurityField.NAME4,
524+
SecurityField.NIO,
522525
httpType);
523526
throw new IllegalArgumentException(message);
524527
}
525528
} else {
526529
// default to security4
527530
builder.put(NetworkModule.HTTP_TYPE_KEY, SecurityField.NAME4);
528-
SecurityNetty4HttpServerTransport.overrideSettings(builder, settings);
531+
SecurityHttpSettings.overrideSettings(builder, settings);
529532
}
530533
builder.put(SecuritySettings.addUserSettings(settings));
531534
return builder.build();
@@ -869,8 +872,14 @@ public Map<String, Supplier<HttpServerTransport>> getHttpTransports(Settings set
869872
if (enabled == false) { // don't register anything if we are not enabled
870873
return Collections.emptyMap();
871874
}
872-
return Collections.singletonMap(SecurityField.NAME4, () -> new SecurityNetty4HttpServerTransport(settings,
873-
networkService, bigArrays, ipFilter.get(), getSslService(), threadPool, xContentRegistry, dispatcher));
875+
876+
Map<String, Supplier<HttpServerTransport>> httpTransports = new HashMap<>();
877+
httpTransports.put(SecurityField.NAME4, () -> new SecurityNetty4HttpServerTransport(settings, networkService, bigArrays,
878+
ipFilter.get(), getSslService(), threadPool, xContentRegistry, dispatcher));
879+
httpTransports.put(SecurityField.NIO, () -> new SecurityNioHttpServerTransport(settings, networkService, bigArrays,
880+
pageCacheRecycler, threadPool, xContentRegistry, dispatcher, ipFilter.get(), getSslService()));
881+
882+
return httpTransports;
874883
}
875884

876885
@Override

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,6 @@
55
*/
66
package org.elasticsearch.xpack.security.rest;
77

8-
import io.netty.channel.Channel;
9-
import io.netty.handler.ssl.SslHandler;
108
import org.apache.logging.log4j.Logger;
119
import org.apache.logging.log4j.message.ParameterizedMessage;
1210
import org.apache.logging.log4j.util.Supplier;
@@ -15,7 +13,6 @@
1513
import org.elasticsearch.common.logging.ESLoggerFactory;
1614
import org.elasticsearch.common.util.concurrent.ThreadContext;
1715
import org.elasticsearch.http.HttpChannel;
18-
import org.elasticsearch.http.netty4.Netty4HttpChannel;
1916
import org.elasticsearch.license.XPackLicenseState;
2017
import org.elasticsearch.rest.BytesRestResponse;
2118
import org.elasticsearch.rest.RestChannel;
@@ -24,7 +21,7 @@
2421
import org.elasticsearch.rest.RestRequest.Method;
2522
import org.elasticsearch.xpack.core.security.rest.RestRequestFilter;
2623
import org.elasticsearch.xpack.security.authc.AuthenticationService;
27-
import org.elasticsearch.xpack.security.transport.ServerTransportFilter;
24+
import org.elasticsearch.xpack.security.transport.SSLEngineUtils;
2825

2926
import java.io.IOException;
3027

@@ -53,10 +50,7 @@ public void handleRequest(RestRequest request, RestChannel channel, NodeClient c
5350
// CORS - allow for preflight unauthenticated OPTIONS request
5451
if (extractClientCertificate) {
5552
HttpChannel httpChannel = request.getHttpChannel();
56-
Channel nettyChannel = ((Netty4HttpChannel) httpChannel).getNettyChannel();
57-
SslHandler handler = nettyChannel.pipeline().get(SslHandler.class);
58-
assert handler != null;
59-
ServerTransportFilter.extractClientCertificates(logger, threadContext, handler.engine(), nettyChannel);
53+
SSLEngineUtils.extractClientCertificates(logger, threadContext, httpChannel);
6054
}
6155
service.authenticate(maybeWrapRestRequest(request), ActionListener.wrap(
6256
authentication -> {

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SSLEngineUtils.java

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.security.transport;
7+
8+
import io.netty.channel.Channel;
9+
import io.netty.handler.ssl.SslHandler;
10+
import org.apache.logging.log4j.Logger;
11+
import org.apache.logging.log4j.message.ParameterizedMessage;
12+
import org.apache.logging.log4j.util.Supplier;
13+
import org.elasticsearch.common.util.concurrent.ThreadContext;
14+
import org.elasticsearch.http.HttpChannel;
15+
import org.elasticsearch.http.netty4.Netty4HttpChannel;
16+
import org.elasticsearch.http.nio.NioHttpChannel;
17+
import org.elasticsearch.nio.SocketChannelContext;
18+
import org.elasticsearch.transport.TcpChannel;
19+
import org.elasticsearch.transport.netty4.Netty4TcpChannel;
20+
import org.elasticsearch.transport.nio.NioTcpChannel;
21+
import org.elasticsearch.xpack.security.authc.pki.PkiRealm;
22+
import org.elasticsearch.xpack.security.transport.nio.SSLChannelContext;
23+
24+
import javax.net.ssl.SSLEngine;
25+
import javax.net.ssl.SSLPeerUnverifiedException;
26+
import java.security.cert.Certificate;
27+
import java.security.cert.X509Certificate;
28+
29+
public class SSLEngineUtils {
30+
31+
private SSLEngineUtils() {}
32+
33+
public static void extractClientCertificates(Logger logger, ThreadContext threadContext, HttpChannel httpChannel) {
34+
SSLEngine sslEngine = getSSLEngine(httpChannel);
35+
extract(logger, threadContext, sslEngine, httpChannel);
36+
}
37+
38+
public static void extractClientCertificates(Logger logger, ThreadContext threadContext, TcpChannel tcpChannel) {
39+
SSLEngine sslEngine = getSSLEngine(tcpChannel);
40+
extract(logger, threadContext, sslEngine, tcpChannel);
41+
}
42+
43+
public static SSLEngine getSSLEngine(HttpChannel httpChannel) {
44+
if (httpChannel instanceof Netty4HttpChannel) {
45+
Channel nettyChannel = ((Netty4HttpChannel) httpChannel).getNettyChannel();
46+
SslHandler handler = nettyChannel.pipeline().get(SslHandler.class);
47+
assert handler != null : "Must have SslHandler";
48+
return handler.engine();
49+
} else if (httpChannel instanceof NioHttpChannel) {
50+
SocketChannelContext context = ((NioHttpChannel) httpChannel).getContext();
51+
assert context instanceof SSLChannelContext : "Must be SSLChannelContext.class, found: " + context.getClass();
52+
return ((SSLChannelContext) context).getSSLEngine();
53+
} else {
54+
throw new AssertionError("Unknown channel class type: " + httpChannel.getClass());
55+
}
56+
}
57+
58+
public static SSLEngine getSSLEngine(TcpChannel tcpChannel) {
59+
if (tcpChannel instanceof Netty4TcpChannel) {
60+
Channel nettyChannel = ((Netty4TcpChannel) tcpChannel).getNettyChannel();
61+
SslHandler handler = nettyChannel.pipeline().get(SslHandler.class);
62+
assert handler != null : "Must have SslHandler";
63+
return handler.engine();
64+
} else if (tcpChannel instanceof NioTcpChannel) {
65+
SocketChannelContext context = ((NioTcpChannel) tcpChannel).getContext();
66+
assert context instanceof SSLChannelContext : "Must be SSLChannelContext.class, found: " + context.getClass();
67+
return ((SSLChannelContext) context).getSSLEngine();
68+
} else {
69+
throw new AssertionError("Unknown channel class type: " + tcpChannel.getClass());
70+
}
71+
}
72+
73+
private static void extract(Logger logger, ThreadContext threadContext, SSLEngine sslEngine, Object channel) {
74+
try {
75+
Certificate[] certs = sslEngine.getSession().getPeerCertificates();
76+
if (certs instanceof X509Certificate[]) {
77+
threadContext.putTransient(PkiRealm.PKI_CERT_HEADER_NAME, certs);
78+
}
79+
} catch (SSLPeerUnverifiedException e) {
80+
// this happens when client authentication is optional and the client does not provide credentials. If client
81+
// authentication was required then this connection should be closed before ever getting into this class
82+
assert sslEngine.getNeedClientAuth() == false;
83+
assert sslEngine.getWantClientAuth();
84+
if (logger.isTraceEnabled()) {
85+
logger.trace(
86+
(Supplier<?>) () -> new ParameterizedMessage(
87+
"SSL Peer did not present a certificate on channel [{}]", channel), e);
88+
} else if (logger.isDebugEnabled()) {
89+
logger.debug("SSL Peer did not present a certificate on channel [{}]", channel);
90+
}
91+
}
92+
}
93+
}

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityHttpExceptionHandler.java

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.security.transport;
7+
8+
import org.apache.logging.log4j.Logger;
9+
import org.apache.logging.log4j.message.ParameterizedMessage;
10+
import org.elasticsearch.common.component.Lifecycle;
11+
import org.elasticsearch.common.network.CloseableChannel;
12+
import org.elasticsearch.http.HttpChannel;
13+
14+
import java.util.function.BiConsumer;
15+
16+
import static org.elasticsearch.xpack.core.security.transport.SSLExceptionHelper.isCloseDuringHandshakeException;
17+
import static org.elasticsearch.xpack.core.security.transport.SSLExceptionHelper.isNotSslRecordException;
18+
import static org.elasticsearch.xpack.core.security.transport.SSLExceptionHelper.isReceivedCertificateUnknownException;
19+
20+
public final class SecurityHttpExceptionHandler implements BiConsumer<HttpChannel, Exception> {
21+
22+
private final Lifecycle lifecycle;
23+
private final Logger logger;
24+
private final BiConsumer<HttpChannel, Exception> fallback;
25+
26+
public SecurityHttpExceptionHandler(Logger logger, Lifecycle lifecycle, BiConsumer<HttpChannel, Exception> fallback) {
27+
this.lifecycle = lifecycle;
28+
this.logger = logger;
29+
this.fallback = fallback;
30+
}
31+
32+
public void accept(HttpChannel channel, Exception e) {
33+
if (!lifecycle.started()) {
34+
return;
35+
}
36+
37+
if (isNotSslRecordException(e)) {
38+
if (logger.isTraceEnabled()) {
39+
logger.trace(new ParameterizedMessage("received plaintext http traffic on a https channel, closing connection {}",
40+
channel), e);
41+
} else {
42+
logger.warn("received plaintext http traffic on a https channel, closing connection {}", channel);
43+
}
44+
CloseableChannel.closeChannel(channel);
45+
} else if (isCloseDuringHandshakeException(e)) {
46+
if (logger.isTraceEnabled()) {
47+
logger.trace(new ParameterizedMessage("connection {} closed during ssl handshake", channel), e);
48+
} else {
49+
logger.warn("connection {} closed during ssl handshake", channel);
50+
}
51+
CloseableChannel.closeChannel(channel);
52+
} else if (isReceivedCertificateUnknownException(e)) {
53+
if (logger.isTraceEnabled()) {
54+
logger.trace(new ParameterizedMessage("http client did not trust server's certificate, closing connection {}",
55+
channel), e);
56+
} else {
57+
logger.warn("http client did not trust this server's certificate, closing connection {}", channel);
58+
}
59+
CloseableChannel.closeChannel(channel);
60+
} else {
61+
fallback.accept(channel, e);
62+
}
63+
}
64+
}

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityHttpSettings.java

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License;
4+
* you may not use this file except in compliance with the Elastic License.
5+
*/
6+
package org.elasticsearch.xpack.security.transport;
7+
8+
import org.elasticsearch.common.settings.Settings;
9+
10+
import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_COMPRESSION;
11+
import static org.elasticsearch.xpack.core.XPackSettings.HTTP_SSL_ENABLED;
12+
13+
public final class SecurityHttpSettings {
14+
15+
private SecurityHttpSettings() {}
16+
17+
public static void overrideSettings(Settings.Builder settingsBuilder, Settings settings) {
18+
if (HTTP_SSL_ENABLED.get(settings) && SETTING_HTTP_COMPRESSION.exists(settings) == false) {
19+
settingsBuilder.put(SETTING_HTTP_COMPRESSION.getKey(), false);
20+
}
21+
}
22+
}

0 commit comments

Comments
 (0)