Skip to content

Commit

Permalink
Merge #3484 into 2.0.0-M4
Browse files Browse the repository at this point in the history
  • Loading branch information
@violetagg
violetagg committed Oct 28, 2024
2 parents 0e75666 + 7134e30 commit 4a87aa4
Show file tree
Hide file tree
Showing 3 changed files with 181 additions and 98 deletions.
66 changes: 42 additions & 24 deletions ...tor-netty5-core/src/main/java/reactor/netty5/channel/MicrometerChannelMetricsHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import io.micrometer.observation.Observation;
import io.netty5.channel.ChannelHandler;
import io.netty5.channel.ChannelHandlerContext;
import io.netty5.handler.ssl.SniCompletionEvent;
import io.netty5.handler.ssl.SslHandler;
import io.netty5.util.concurrent.Future;
import reactor.netty5.ReactorNetty;
Expand Down Expand Up @@ -173,6 +174,8 @@ static final class TlsMetricsHandler extends Observation.Context
final MicrometerChannelMetricsRecorder recorder;
final SocketAddress remoteAddress;
final String type;

boolean listenerAdded;
Observation observation;

// remote address and status are not known beforehand
Expand All @@ -192,30 +195,7 @@ static final class TlsMetricsHandler extends Observation.Context
@Override
@SuppressWarnings("try")
public void channelActive(ChannelHandlerContext ctx) {
SocketAddress rAddr = remoteAddress != null ? remoteAddress : ctx.channel().remoteAddress();
if (rAddr instanceof InetSocketAddress address) {
this.netPeerName = address.getHostString();
this.netPeerPort = address.getPort() + "";
}
else {
this.netPeerName = rAddr.toString();
this.netPeerPort = "";
}
observation = Observation.createNotStarted(recorder.name() + TLS_HANDSHAKE_TIME, this, OBSERVATION_REGISTRY);
parentContextView = updateChannelContext(ctx.channel(), observation);
observation.start();
ctx.pipeline()
.get(SslHandler.class)
.handshakeFuture()
.addListener(f -> {
ctx.pipeline().remove(this);
status = f.isSuccess() ? SUCCESS : ERROR;
observation.stop();

ReactorNetty.setChannelContext(ctx.channel(), parentContextView);
parentContextView = null;
});

addListener(ctx);
ctx.fireChannelActive();
}

Expand Down Expand Up @@ -247,9 +227,47 @@ public KeyValues getLowCardinalityKeyValues() {
}
}

@Override
public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof SniCompletionEvent) {
addListener(ctx);
}
ctx.fireChannelInboundEvent(evt);
}

@Override
public Timer getTimer() {
return recorder.getTlsHandshakeTimer(getName(), netPeerName + ':' + netPeerPort, proxyAddress == null ? NA : proxyAddress, status);
}

private void addListener(ChannelHandlerContext ctx) {
if (!listenerAdded) {
SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
if (sslHandler != null) {
listenerAdded = true;
SocketAddress rAddr = remoteAddress != null ? remoteAddress : ctx.channel().remoteAddress();
if (rAddr instanceof InetSocketAddress address) {
this.netPeerName = address.getHostString();
this.netPeerPort = address.getPort() + "";
}
else {
this.netPeerName = rAddr.toString();
this.netPeerPort = "";
}
observation = Observation.createNotStarted(recorder.name() + TLS_HANDSHAKE_TIME, this, OBSERVATION_REGISTRY);
parentContextView = updateChannelContext(ctx.channel(), observation);
observation.start();
sslHandler.handshakeFuture()
.addListener(f -> {
ctx.pipeline().remove(this);
status = f.isSuccess() ? SUCCESS : ERROR;
observation.stop();

ReactorNetty.setChannelContext(ctx.channel(), parentContextView);
parentContextView = null;
});
}
}
}
}
}
60 changes: 38 additions & 22 deletions reactor-netty5-http/src/main/java/reactor/netty5/http/server/HttpServerConfig.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import io.netty5.handler.codec.http2.Http2StreamFrameToHttpObjectCodec;
import io.netty5.handler.logging.LogLevel;
import io.netty5.handler.logging.LoggingHandler;
import io.netty5.handler.ssl.AbstractSniHandler;
import io.netty5.handler.ssl.ApplicationProtocolNames;
import io.netty5.handler.ssl.ApplicationProtocolNegotiationHandler;
import io.netty5.handler.timeout.ReadTimeoutHandler;
Expand Down Expand Up @@ -1105,9 +1106,14 @@ static final class H2OrHttp11Codec extends ApplicationProtocolNegotiationHandler
final ChannelOperations.OnSetup opsFactory;
final Duration readTimeout;
final Duration requestTimeout;
final boolean supportOnlyHttp2;
final Function<String, String> uriTagValue;

H2OrHttp11Codec(HttpServerChannelInitializer initializer, ConnectionObserver listener) {
this(initializer, listener, false);
}

H2OrHttp11Codec(HttpServerChannelInitializer initializer, ConnectionObserver listener, boolean supportOnlyHttp2) {
super(ApplicationProtocolNames.HTTP_1_1);
this.accessLogEnabled = initializer.accessLogEnabled;
this.accessLog = initializer.accessLog;
Expand All @@ -1128,6 +1134,7 @@ static final class H2OrHttp11Codec extends ApplicationProtocolNegotiationHandler
this.opsFactory = initializer.opsFactory;
this.readTimeout = initializer.readTimeout;
this.requestTimeout = initializer.requestTimeout;
this.supportOnlyHttp2 = supportOnlyHttp2;
this.uriTagValue = initializer.uriTagValue;
}

Expand All @@ -1146,7 +1153,7 @@ protected void configurePipeline(ChannelHandlerContext ctx, String protocol) {
return;
}

if (ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
if (!supportOnlyHttp2 && ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
configureHttp11Pipeline(p, accessLogEnabled, accessLog, compressPredicate, true,
decoder, formDecoderProvider, forwardedHeaderHandler, httpMessageLogFactory, idleTimeout, listener, mapHandle,
maxKeepAliveRequests, methodTagValue, metricsRecorder, minCompressionSize, readTimeout, requestTimeout, uriTagValue);
Expand Down Expand Up @@ -1258,27 +1265,36 @@ else if ((protocols & h11) == h11) {
uriTagValue);
}
else if ((protocols & h2) == h2) {
configureH2Pipeline(
channel.pipeline(),
accessLogEnabled,
accessLog,
compressPredicate(compressPredicate, minCompressionSize),
enableGracefulShutdown,
formDecoderProvider,
forwardedHeaderHandler,
http2SettingsSpec,
httpMessageLogFactory,
idleTimeout,
observer,
mapHandle,
methodTagValue,
metricsRecorder,
minCompressionSize,
opsFactory,
readTimeout,
requestTimeout,
uriTagValue,
decoder.validateHeaders());
ChannelHandler sslHandler = channel.pipeline().get(NettyPipeline.SslHandler);
if (sslHandler instanceof AbstractSniHandler) {
channel.pipeline()
.addBefore(NettyPipeline.ReactiveBridge,
NettyPipeline.H2OrHttp11Codec,
new H2OrHttp11Codec(this, observer, true));
}
else {
configureH2Pipeline(
channel.pipeline(),
accessLogEnabled,
accessLog,
compressPredicate(compressPredicate, minCompressionSize),
enableGracefulShutdown,
formDecoderProvider,
forwardedHeaderHandler,
http2SettingsSpec,
httpMessageLogFactory,
idleTimeout,
observer,
mapHandle,
methodTagValue,
metricsRecorder,
minCompressionSize,
opsFactory,
readTimeout,
requestTimeout,
uriTagValue,
decoder.validateHeaders());
}
}
}
else {
Expand Down
153 changes: 101 additions & 52 deletions reactor-netty5-http/src/test/java/reactor/netty5/http/server/HttpServerTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2022,12 +2022,18 @@ void testHang() {
}

@Test
void testSniSupport() throws Exception {
void testSniSupportHttp11() throws Exception {
doTestSniSupport(Function.identity(), Function.identity());
}

@ParameterizedTest
@MethodSource("h2CompatibleCombinations")
void testSniSupportHttp2(HttpProtocol[] serverProtocols, HttpProtocol[] clientProtocols) throws Exception {
doTestSniSupport(server -> server.protocol(serverProtocols), client -> client.protocol(clientProtocols));
}

@Test
void testIssue3022() throws Exception {
void testIssue3022Http11() throws Exception {
TestHttpClientMetricsRecorder clientMetricsRecorder = new TestHttpClientMetricsRecorder();
TestHttpServerMetricsRecorder serverMetricsRecorder = new TestHttpServerMetricsRecorder();
doTestSniSupport(server -> server.metrics(true, () -> serverMetricsRecorder, Function.identity()),
Expand All @@ -2036,39 +2042,67 @@ void testIssue3022() throws Exception {
assertThat(serverMetricsRecorder.tlsHandshakeTime).isNotNull().isGreaterThan(Duration.ZERO);
}

@ParameterizedTest
@MethodSource("h2CompatibleCombinations")
void testIssue3022Http2(HttpProtocol[] serverProtocols, HttpProtocol[] clientProtocols) throws Exception {
TestHttpClientMetricsRecorder clientMetricsRecorder = new TestHttpClientMetricsRecorder();
TestHttpServerMetricsRecorder serverMetricsRecorder = new TestHttpServerMetricsRecorder();
doTestSniSupport(server -> server.protocol(serverProtocols).metrics(true, () -> serverMetricsRecorder, Function.identity()),
client -> client.protocol(clientProtocols).metrics(true, () -> clientMetricsRecorder, Function.identity()));
assertThat(clientMetricsRecorder.tlsHandshakeTime).isNotNull().isGreaterThan(Duration.ZERO);
assertThat(serverMetricsRecorder.tlsHandshakeTime).isNotNull().isGreaterThan(Duration.ZERO);
}

@Test
void testIssue3473Http11() throws Exception {
doTestSniSupport(server -> server.metrics(true, Function.identity()),
client -> client.metrics(true, Function.identity()));
}

@ParameterizedTest
@MethodSource("h2CompatibleCombinations")
void testIssue3473Http2(HttpProtocol[] serverProtocols, HttpProtocol[] clientProtocols) throws Exception {
doTestSniSupport(server -> server.protocol(serverProtocols).metrics(true, Function.identity()),
client -> client.protocol(clientProtocols).metrics(true, Function.identity()));
}

@SuppressWarnings("deprecation")
private void doTestSniSupport(Function<HttpServer, HttpServer> serverCustomizer,
Function<HttpClient, HttpClient> clientCustomizer) throws Exception {
SelfSignedCertificate defaultCert = new SelfSignedCertificate("default");
Http11SslContextSpec defaultSslContextBuilder =
Http11SslContextSpec.forServer(defaultCert.certificate(), defaultCert.privateKey());

SelfSignedCertificate testCert = new SelfSignedCertificate("test.com");
Http11SslContextSpec testSslContextBuilder =
Http11SslContextSpec.forServer(testCert.certificate(), testCert.privateKey());

Http11SslContextSpec clientSslContextBuilder =
Http11SslContextSpec.forClient()
.configure(builder -> builder.trustManager(InsecureTrustManagerFactory.INSTANCE));

AtomicReference<String> hostname = new AtomicReference<>();
HttpServer server = serverCustomizer.apply(createServer());

boolean isH2 = (server.configuration()._protocols & HttpServerConfig.h2) == HttpServerConfig.h2;
SslProvider.ProtocolSslContextSpec defaultSslContextBuilder = isH2 ?
Http2SslContextSpec.forServer(defaultCert.certificate(), defaultCert.privateKey()) :
Http11SslContextSpec.forServer(defaultCert.certificate(), defaultCert.privateKey());
SslProvider.ProtocolSslContextSpec testSslContextBuilder = isH2 ?
Http2SslContextSpec.forServer(testCert.certificate(), testCert.privateKey()) :
Http11SslContextSpec.forServer(testCert.certificate(), testCert.privateKey());

disposableServer =
serverCustomizer.apply(createServer())
.secure(spec -> spec.sslContext(defaultSslContextBuilder)
.addSniMapping("*.test.com", domainSpec -> domainSpec.sslContext(testSslContextBuilder)))
.doOnChannelInit((obs, channel, remoteAddress) ->
channel.pipeline()
.addAfter(NettyPipeline.SslHandler, "test", new ChannelHandlerAdapter() {
@Override
public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof SniCompletionEvent sniCompletionEvent) {
hostname.set(sniCompletionEvent.hostname());
}
ctx.fireChannelInboundEvent(evt);
server.secure(spec -> spec.sslContext(defaultSslContextBuilder)
.addSniMapping("*.test.com", domainSpec -> domainSpec.sslContext(testSslContextBuilder)))
.doOnChannelInit((obs, channel, remoteAddress) ->
channel.pipeline()
.addAfter(NettyPipeline.SslHandler, "test", new ChannelHandlerAdapter() {
@Override
public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof SniCompletionEvent) {
hostname.set(((SniCompletionEvent) evt).hostname());
}
}))
.handle((req, res) -> res.sendString(Mono.just("testSniSupport")))
.bindNow();
ctx.fireChannelInboundEvent(evt);
}
}))
.handle((req, res) -> res.sendString(Mono.just("testSniSupport")))
.bindNow();

SslProvider.ProtocolSslContextSpec clientSslContextBuilder = isH2 ?
Http2SslContextSpec.forClient().configure(builder -> builder.trustManager(InsecureTrustManagerFactory.INSTANCE)) :
Http11SslContextSpec.forClient().configure(builder -> builder.trustManager(InsecureTrustManagerFactory.INSTANCE));

clientCustomizer.apply(createClient(disposableServer::address))
.secure(spec -> spec.sslContext(clientSslContextBuilder)
Expand All @@ -2084,41 +2118,56 @@ public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
}

@Test
void testSniSupportAsyncMappingsHttp11() throws Exception {
doTestSniSupportAsyncMappings(Function.identity(), Function.identity());
}

@ParameterizedTest
@MethodSource("h2CompatibleCombinations")
void testSniSupportAsyncMappingsHttp2(HttpProtocol[] serverProtocols, HttpProtocol[] clientProtocols) throws Exception {
doTestSniSupportAsyncMappings(server -> server.protocol(serverProtocols), client -> client.protocol(clientProtocols));
}

@SuppressWarnings("deprecation")
void testSniSupportAsyncMappings() throws Exception {
private void doTestSniSupportAsyncMappings(Function<HttpServer, HttpServer> serverCustomizer,
Function<HttpClient, HttpClient> clientCustomizer) throws Exception {
SelfSignedCertificate defaultCert = new SelfSignedCertificate("default");
Http11SslContextSpec defaultSslContextBuilder =
Http11SslContextSpec.forServer(defaultCert.certificate(), defaultCert.privateKey());

SelfSignedCertificate testCert = new SelfSignedCertificate("test.com");
Http11SslContextSpec testSslContextBuilder =

AtomicReference<String> hostname = new AtomicReference<>();
HttpServer server = serverCustomizer.apply(createServer());

boolean isH2 = (server.configuration()._protocols & HttpServerConfig.h2) == HttpServerConfig.h2;
SslProvider.ProtocolSslContextSpec defaultSslContextBuilder = isH2 ?
Http2SslContextSpec.forServer(defaultCert.certificate(), defaultCert.privateKey()) :
Http11SslContextSpec.forServer(defaultCert.certificate(), defaultCert.privateKey());
SslProvider.ProtocolSslContextSpec testSslContextBuilder = isH2 ?
Http2SslContextSpec.forServer(testCert.certificate(), testCert.privateKey()) :
Http11SslContextSpec.forServer(testCert.certificate(), testCert.privateKey());
SslProvider testSslProvider = SslProvider.builder().sslContext(testSslContextBuilder).build();

Http11SslContextSpec clientSslContextBuilder =
Http11SslContextSpec.forClient()
.configure(builder -> builder.trustManager(InsecureTrustManagerFactory.INSTANCE));

AtomicReference<String> hostname = new AtomicReference<>();
disposableServer =
createServer()
.secure(spec -> spec.sslContext(defaultSslContextBuilder)
server.secure(spec -> spec.sslContext(defaultSslContextBuilder)
.setSniAsyncMappings((input, promise) -> promise.setSuccess(testSslProvider).asFuture()))
.doOnChannelInit((obs, channel, remoteAddress) ->
channel.pipeline()
.addAfter(NettyPipeline.SslHandler, "test", new ChannelHandlerAdapter() {
@Override
public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof SniCompletionEvent sniCompletionEvent) {
hostname.set(sniCompletionEvent.hostname());
}
ctx.fireChannelInboundEvent(evt);
}
}))
.handle((req, res) -> res.sendString(Mono.just("testSniSupport")))
.bindNow();
.doOnChannelInit((obs, channel, remoteAddress) ->
channel.pipeline()
.addAfter(NettyPipeline.SslHandler, "test", new ChannelHandlerAdapter() {
@Override
public void channelInboundEvent(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof SniCompletionEvent) {
hostname.set(((SniCompletionEvent) evt).hostname());
}
ctx.fireChannelInboundEvent(evt);
}
}))
.handle((req, res) -> res.sendString(Mono.just("testSniSupport")))
.bindNow();

createClient(disposableServer::address)
SslProvider.ProtocolSslContextSpec clientSslContextBuilder = isH2 ?
Http2SslContextSpec.forClient().configure(builder -> builder.trustManager(InsecureTrustManagerFactory.INSTANCE)) :
Http11SslContextSpec.forClient().configure(builder -> builder.trustManager(InsecureTrustManagerFactory.INSTANCE));

clientCustomizer.apply(createClient(disposableServer::address))
.secure(spec -> spec.sslContext(clientSslContextBuilder)
.serverNames(new SNIHostName("test.com")))
.get()
Expand Down

0 comments on commit 4a87aa4

Please sign in to comment.