Adding Spring security to WebFlux flow fails with blocking errors

[ekazakas]

Hello, I have a very simple flow:

@Configuration
public class LiveMessagesFlow {
	@Bean
	public IntegrationFlow metricGamingInboundFlow() {
		return IntegrationFlows
				.from(
						WebFlux.inboundGateway("/api/live/message")
								.requestMapping(mapping -> mapping
										.consumes(MediaType.APPLICATION_JSON_VALUE)
										.produces(MediaType.APPLICATION_JSON_VALUE)
								)
				)
				.log()
                                .handle((p, h) -> Mono.just("{\"status\":\"OK\"}"))
				.get();
	}
}

Without adding Spring Security, this works properly, but after adding the following:

@EnableWebFluxSecurity
@Configuration
public class SecurityConfiguration {
	private static final String ROLE_USER = "USER";

	@Bean
	public SecurityWebFilterChain springSecurityFilterChain(ServerHttpSecurity http) {
		http.csrf()
				.disable()
				.authorizeExchange()
				.matchers(EndpointRequest.toAnyEndpoint())
				.permitAll()
				.anyExchange()
				.hasRole(ROLE_USER)
				.and()
				.httpBasic();

		return http.build();
	}

	@Bean
	public ReactiveUserDetailsService userDetailsService(SecurityProperties properties) {
		return new MapReactiveUserDetailsService(Arrays.asList(buildUserDetails(properties.getUser())));
	}

	private UserDetails buildUserDetails(SecurityProperties.User user) {
		return User.withUsername(user.getName())
				.password(user.getPassword())
				.roles(user.getRoles().stream().toArray(String[]::new))
				.build();
	}
}

I get the following exception when trying to call the endpoint:

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-server-epoll-7
	at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:77) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.Mono.block(Mono.java:1175) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint.buildMessage(WebFluxInboundEndpoint.java:310) ~[spring-integration-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]
	at org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint.lambda$doHandle$3(WebFluxInboundEndpoint.java:155) ~[spring-integration-webflux-5.0.5.RELEASE.jar:5.0.5.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:108) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:198) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:104) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1080) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.MonoCollectList$MonoBufferAllSubscriber.onComplete(MonoCollectList.java:117) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:130) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:245) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:130) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.ipc.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:377) ~[reactor-netty-0.7.7.RELEASE.jar:0.7.7.RELEASE]
	at reactor.ipc.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:202) ~[reactor-netty-0.7.7.RELEASE.jar:0.7.7.RELEASE]
	at reactor.ipc.netty.channel.FluxReceive.request(FluxReceive.java:110) ~[reactor-netty-0.7.7.RELEASE.jar:0.7.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:149) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.request(FluxPeek.java:130) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:149) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.MonoCollectList$MonoBufferAllSubscriber.onSubscribe(MonoCollectList.java:90) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:86) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onSubscribe(FluxPeek.java:163) ~[reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.core.publisher.FluxMap$MapSubscriber.onSubscribe(FluxMap.java:86) [reactor-core-3.1.7.RELEASE.jar:3.1.7.RELEASE]
	at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:273) ~[reactor-netty-0.7.7.RELEASE.jar:0.7.7.RELEASE]
	at reactor.ipc.netty.channel.FluxReceive.lambda$subscribe$2(FluxReceive.java:127) ~[reactor-netty-0.7.7.RELEASE.jar:0.7.7.RELEASE]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) ~[netty-common-4.1.24.Final.jar:4.1.24.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.24.Final.jar:4.1.24.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404) ~[netty-common-4.1.24.Final.jar:4.1.24.Final]
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:309) ~[netty-transport-native-epoll-4.1.24.Final-linux-x86_64.jar:4.1.24.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884) ~[netty-common-4.1.24.Final.jar:4.1.24.Final]
	at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_171]

Spring dependencies that I currently use:

org.springframework.boot:spring-boot-starter-integration:2.0.2.RELEASE
org.springframework.boot:spring-boot-starter-webflux:2.0.2.RELEASE
org.springframework.boot:spring-boot-starter-actuator:2.0.2.RELEASE
org.springframework.boot:spring-boot-starter-security:2.0.2.RELEASE
org.springframework.boot:spring-boot-starter-amqp:2.0.2.RELEASE
org.springframework.integration:spring-integration-webflux:5.0.5.RELEASE
org.springframework.integration:spring-integration-amqp:5.0.5.RELEASE

It seems there is an issue with the following line in WebFluxInboundEndpoint:

....setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL, exchange.getPrincipal().block())

Can someon provide some insight on this?

[artembilan]

I think this really just a bug.
Originally I thought it isn't going to be a pain, but now since Reactor has such a protection, I think we need to fix it immediately as just a Mono - not its block().

We would need a note in the Reference Manual and some Migration Guide bullet.

Thank you for sharing it !

[ekazakas]

@artembilan Thanks Artem! Good to get some feedback. I assume there is no workaround for this at the moment and I will have to wait for a new release.

Edit: Tried to go back to Spring Boot 2.0.0. There my code works properly. So it seems this is a regression due to increased protection on Reactor side.

[denisrobert]

How could this get through? No one would dream of using any sort of web endpoint without security protection in real code, so it's hard to understand how this bug can even be. How could reactor add such breaking changes in a point release (seems the constraints were added on the sly in reactor 3.1.6).

This is a major showstopper to using the WebFlux endpoints. And since you can't mix WebFlux and MVC, it prevents using spring-integration web endpoints in a WebFlux-based application.

[artembilan]

@denisrobert ,

Thank you for your feedback!
Well, we had a test-case with the WebFlux and, respectively, Reactor, but as you see we leaked a Security bit there. That's how it got through. The test-case is a simple unit test against WebFlux. We don't test there the Security at all just because we don't produce any functionality around that part. It is really not an integration test to cover everything. That's not a responsibility of this part of the project.

Now I added such a Security configuration for the WebFlux part as well. Just because we have this incident and that really wasn't obvious from the beginning that we could fail with the security.

The Reactor story is a bit different and even if it is some kind of breaking change from their side, that is really our fault do not follow non-blocking rules in such a WebFlux scenario.

I think we are going to release version 5.0.6 the next week just in time for the Boot 1.5.14, so that is not so long period that you need to downgrade Reactor version explicitly in your project.

Sometimes thing are so clear and obvious that we have everything covered from the day first, but sometimes we don't have enough knowledge or are in rash that may miss something not so obvious. More over we always try to be responsive and opened for any community feedback, so we close our bugs as quick as possible.

Does it all make sense to you?

[ekazakas]

People need to understand that Spring Boot 2.0 is only a couple of months old, it's still brand new and fresh tech., I don't expect it to be 100% stable. It's an open source project and if you find any issues, report them ASAP, provide examples, make code contributions. Allow it to mature a bit. Also, in my previous comment, I confirmed that in 2.0.0, it's working perfectly fine, because slightly older Reactor doesn't have this "safety" check to ensure that running thread is non-blocking, this was broken in 2.0.1, because Reactor dependency was updated and the WebFlux endpoint has some blocking logic that blocks the running thread. I expect that this fix will roll out with 2.0.3 release.