Polish "Add SSL support to RSocketServer"

See gh-19399

Author: bclozel

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/rsocket/RSocketProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactory.java

@@ -151,41 +151,27 @@ private ServerTransport<CloseableChannel> createTransport() {
 	}
 
 	private ServerTransport<CloseableChannel> createWebSocketTransport() {
-		HttpServer httpServer;
+		HttpServer httpServer = HttpServer.create();
 		if (this.resourceFactory != null) {
-			httpServer = HttpServer.create().runOn(this.resourceFactory.getLoopResources())
-					.bindAddress(this::getListenAddress);
+			httpServer = httpServer.runOn(this.resourceFactory.getLoopResources());
 		}
-		else {
-			InetSocketAddress listenAddress = this.getListenAddress();
-			httpServer = HttpServer.create().host(listenAddress.getHostName()).port(listenAddress.getPort());
-		}
-
 		if (this.ssl != null && this.ssl.isEnabled()) {
 			SslServerCustomizer sslServerCustomizer = new SslServerCustomizer(this.ssl, null, this.sslStoreProvider);
 			httpServer = sslServerCustomizer.apply(httpServer);
 		}
-
-		return WebsocketServerTransport.create(httpServer);
+		return WebsocketServerTransport.create(httpServer.bindAddress(this::getListenAddress));
 	}
 
 	private ServerTransport<CloseableChannel> createTcpTransport() {
-		TcpServer tcpServer;
+		TcpServer tcpServer = TcpServer.create();
 		if (this.resourceFactory != null) {
-			tcpServer = TcpServer.create().runOn(this.resourceFactory.getLoopResources())
-					.bindAddress(this::getListenAddress);
+			tcpServer = tcpServer.runOn(this.resourceFactory.getLoopResources());
 		}
-		else {
-			InetSocketAddress listenAddress = this.getListenAddress();
-			tcpServer = TcpServer.create().host(listenAddress.getHostName()).port(listenAddress.getPort());
-		}
-
 		if (this.ssl != null && this.ssl.isEnabled()) {
 			TcpSslServerCustomizer sslServerCustomizer = new TcpSslServerCustomizer(this.ssl, this.sslStoreProvider);
 			tcpServer = sslServerCustomizer.apply(tcpServer);
 		}
-
-		return TcpServerTransport.create(tcpServer);
+		return TcpServerTransport.create(tcpServer.bindAddress(this::getListenAddress));
 	}
 
 	private InetSocketAddress getListenAddress() {
@@ -201,9 +187,6 @@ private TcpSslServerCustomizer(Ssl ssl, SslStoreProvider sslStoreProvider) {
 			super(ssl, null, sslStoreProvider);
 		}
 
-		// This does not override the apply in parent - currently just leveraging the
-		// parent for its "getContextBuilder()" method. This should be refactored when
-		// we add the concept of http/tcp customizers for RSocket.
 		private TcpServer apply(TcpServer server) {
 			try {
 				return server.secure((contextSpec) -> contextSpec.sslContext(getContextBuilder()));

spring-boot-project/spring-boot/src/main/java/org/springframework/boot/rsocket/server/ConfigurableRSocketServerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

spring-boot-project/spring-boot/src/test/java/org/springframework/boot/rsocket/netty/NettyRSocketServerFactoryTests.java

@@ -18,7 +18,6 @@
 
 import java.net.InetSocketAddress;
 import java.nio.channels.ClosedChannelException;
-import java.time.Duration;
 import java.util.Arrays;
 import java.util.concurrent.Callable;
 
@@ -38,12 +37,13 @@
 import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
 import reactor.core.publisher.Mono;
+import reactor.netty.http.client.HttpClient;
 import reactor.netty.tcp.TcpClient;
 import reactor.test.StepVerifier;
 
 import org.springframework.boot.rsocket.server.RSocketServer;
-import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
 import org.springframework.boot.rsocket.server.RSocketServer.Transport;
+import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
 import org.springframework.boot.web.server.Ssl;
 import org.springframework.core.codec.CharSequenceEncoder;
 import org.springframework.core.codec.StringDecoder;
@@ -55,7 +55,6 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.will;
 import static org.mockito.Mockito.inOrder;
@@ -74,10 +73,11 @@ class NettyRSocketServerFactoryTests {
 
 	private RSocketRequester requester;
 
-	private static final Duration TIMEOUT = Duration.ofSeconds(3);
-
 	@AfterEach
 	void tearDown() {
+		if (this.requester != null) {
+			this.requester.rsocketClient().dispose();
+		}
 		if (this.server != null) {
 			try {
 				this.server.stop();
@@ -86,9 +86,6 @@ void tearDown() {
 				// Ignore
 			}
 		}
-		if (this.requester != null) {
-			this.requester.rsocketClient().dispose();
-		}
 	}
 
 	private NettyRSocketServerFactory getFactory() {
@@ -105,11 +102,9 @@ void specificPort() {
 			this.server.start();
 			return port;
 		});
-		this.requester = createRSocketTcpClient(false);
-		String payload = "test payload";
-		String response = this.requester.route("test").data(payload).retrieveMono(String.class).block(TIMEOUT);
+		this.requester = createRSocketTcpClient();
 		assertThat(this.server.address().getPort()).isEqualTo(specificPort);
-		assertThat(response).isEqualTo(payload);
+		checkEchoRequest();
 	}
 
 	@Test
@@ -118,10 +113,8 @@ void websocketTransport() {
 		factory.setTransport(RSocketServer.Transport.WEBSOCKET);
 		this.server = factory.create(new EchoRequestResponseAcceptor());
 		this.server.start();
-		this.requester = createRSocketWebSocketClient(false);
-		String payload = "test payload";
-		String response = this.requester.route("test").data(payload).retrieveMono(String.class).block(TIMEOUT);
-		assertThat(response).isEqualTo(payload);
+		this.requester = createRSocketWebSocketClient();
+		checkEchoRequest();
 	}
 
 	@Test
@@ -133,10 +126,8 @@ void websocketTransportWithReactorResource() {
 		factory.setResourceFactory(resourceFactory);
 		this.server = factory.create(new EchoRequestResponseAcceptor());
 		this.server.start();
-		this.requester = createRSocketWebSocketClient(false);
-		String payload = "test payload";
-		String response = this.requester.route("test").data(payload).retrieveMono(String.class).block(TIMEOUT);
-		assertThat(response).isEqualTo(payload);
+		this.requester = createRSocketWebSocketClient();
+		checkEchoRequest();
 	}
 
 	@Test
@@ -176,6 +167,12 @@ void websocketTransportBasicSslFromFileSystem() {
 		testBasicSslWithKeyStore("src/test/resources/test.jks", "password", Transport.WEBSOCKET);
 	}
 
+	private void checkEchoRequest() {
+		String payload = "test payload";
+		Mono<String> response = this.requester.route("test").data(payload).retrieveMono(String.class);
+		StepVerifier.create(response).expectNext(payload).verifyComplete();
+	}
+
 	private void testBasicSslWithKeyStore(String keyStore, String keyPassword, Transport transport) {
 		NettyRSocketServerFactory factory = getFactory();
 		factory.setTransport(transport);
@@ -185,11 +182,9 @@ private void testBasicSslWithKeyStore(String keyStore, String keyPassword, Trans
 		factory.setSsl(ssl);
 		this.server = factory.create(new EchoRequestResponseAcceptor());
 		this.server.start();
-		this.requester = (transport == Transport.TCP) ? createRSocketTcpClient(true)
-				: createRSocketWebSocketClient(true);
-		String payload = "test payload";
-		Mono<String> responseMono = this.requester.route("test").data(payload).retrieveMono(String.class);
-		StepVerifier.create(responseMono).expectNext(payload).verifyComplete();
+		this.requester = (transport == Transport.TCP) ? createSecureRSocketTcpClient()
+				: createSecureRSocketWebSocketClient();
+		checkEchoRequest();
 	}
 
 	@Test
@@ -202,48 +197,54 @@ void tcpTransportSslRejectsInsecureClient() {
 		factory.setSsl(ssl);
 		this.server = factory.create(new EchoRequestResponseAcceptor());
 		this.server.start();
-		this.requester = createRSocketTcpClient(false);
+		this.requester = createRSocketTcpClient();
 		String payload = "test payload";
 		Mono<String> responseMono = this.requester.route("test").data(payload).retrieveMono(String.class);
 		StepVerifier.create(responseMono)
 				.verifyErrorSatisfies((ex) -> assertThatExceptionOfType(ClosedChannelException.class));
 	}
 
-	@Test
-	void websocketTransportSslRejectsInsecureClient() {
-		NettyRSocketServerFactory factory = getFactory();
-		factory.setTransport(Transport.WEBSOCKET);
-		Ssl ssl = new Ssl();
-		ssl.setKeyStore("classpath:test.jks");
-		ssl.setKeyPassword("password");
-		factory.setSsl(ssl);
-		this.server = factory.create(new EchoRequestResponseAcceptor());
-		this.server.start();
-		// For WebSocket, the SSL failure results in a hang on the initial connect call
-		assertThatThrownBy(() -> createRSocketWebSocketClient(false)).isInstanceOf(IllegalStateException.class)
-				.hasStackTraceContaining("Timeout on blocking read");
+	private RSocketRequester createRSocketTcpClient() {
+		return createRSocketRequesterBuilder().transport(TcpClientTransport.create(createTcpClient()));
+	}
+
+	private RSocketRequester createRSocketWebSocketClient() {
+		return createRSocketRequesterBuilder().transport(WebsocketClientTransport.create(createHttpClient(), "/"));
 	}
 
-	private RSocketRequester createRSocketTcpClient(boolean ssl) {
-		TcpClient tcpClient = createTcpClient(ssl);
-		return createRSocketRequesterBuilder().connect(TcpClientTransport.create(tcpClient)).block(TIMEOUT);
+	private RSocketRequester createSecureRSocketTcpClient() {
+		return createRSocketRequesterBuilder().transport(TcpClientTransport.create(createSecureTcpClient()));
 	}
 
-	private RSocketRequester createRSocketWebSocketClient(boolean ssl) {
-		TcpClient tcpClient = createTcpClient(ssl);
-		return createRSocketRequesterBuilder().connect(WebsocketClientTransport.create(tcpClient)).block(TIMEOUT);
+	private RSocketRequester createSecureRSocketWebSocketClient() {
+		return createRSocketRequesterBuilder()
+				.transport(WebsocketClientTransport.create(createSecureHttpClient(), "/"));
 	}
 
-	private TcpClient createTcpClient(boolean ssl) {
+	private HttpClient createSecureHttpClient() {
+		HttpClient httpClient = createHttpClient();
+		SslContextBuilder builder = SslContextBuilder.forClient().sslProvider(SslProvider.JDK)
+				.trustManager(InsecureTrustManagerFactory.INSTANCE);
+		return httpClient.secure((spec) -> spec.sslContext(builder));
+	}
+
+	private HttpClient createHttpClient() {
 		Assertions.assertThat(this.server).isNotNull();
 		InetSocketAddress address = this.server.address();
-		TcpClient tcpClient = TcpClient.create().host(address.getHostName()).port(address.getPort());
-		if (ssl) {
-			SslContextBuilder builder = SslContextBuilder.forClient().sslProvider(SslProvider.JDK)
-					.trustManager(InsecureTrustManagerFactory.INSTANCE);
-			tcpClient = tcpClient.secure((spec) -> spec.sslContext(builder));
-		}
-		return tcpClient;
+		return HttpClient.create().host(address.getHostName()).port(address.getPort());
+	}
+
+	private TcpClient createSecureTcpClient() {
+		TcpClient tcpClient = createTcpClient();
+		SslContextBuilder builder = SslContextBuilder.forClient().sslProvider(SslProvider.JDK)
+				.trustManager(InsecureTrustManagerFactory.INSTANCE);
+		return tcpClient.secure((spec) -> spec.sslContext(builder));
+	}
+
+	private TcpClient createTcpClient() {
+		Assertions.assertThat(this.server).isNotNull();
+		InetSocketAddress address = this.server.address();
+		return TcpClient.create().host(address.getHostName()).port(address.getPort());
 	}
 
 	private RSocketRequester.Builder createRSocketRequesterBuilder() {