java.lang.IllegalStateException: Only one connection receive subscriber allowed.

[hfgbarrigas]

Summary

Trying oauth2 client_credentials integration with webclient but getting the IllegalStateException.

Configuration

Using a custom provider. Here's the configuration:

spring:
  security:
    oauth2:
      client:
        registration:
          iam:
            client-id: my_client_id
            client-secret: my_secret
            provider: as
            clientAuthenticationMethod: basic
            authorizationGrantType: client_credentials
        provider:
          as:
            token-uri: https://my_authorization_server/as/oauth/token

My web security is as follows:

http
                .authorizeExchange()
                .pathMatchers(resourceServerProperties.getPublicEndpoints().toArray(new String[]{}))
                .permitAll()
                .anyExchange().authenticated()
                .and()
                .csrf()
                .disable()
                .oauth2ResourceServer()
                .jwt()
                .jwtDecoder(nimbusReactiveJwtDecoder)
                .and()
                .and()
                .oauth2Client()

Configuring webclient as follows:

ServerOAuth2AuthorizedClientExchangeFilterFunction oauth = new ServerOAuth2AuthorizedClientExchangeFilterFunction(clientRegistrationRepo,
                authorizedClientRepo);

        oauth.setDefaultClientRegistrationId(iamReactiveClientProperties.getDefaultClientRegistrationName());
        oauth.setDefaultOAuth2AuthorizedClient(true);

        return WebClient
                .builder()
                .exchangeStrategies(ExchangeStrategies
                        .builder()
                        .codecs(clientDefaultCodecsConfigurer -> {
                            clientDefaultCodecsConfigurer.defaultCodecs().jackson2JsonEncoder(new Jackson2JsonEncoder(objectMapper, MediaType.APPLICATION_JSON));
                            clientDefaultCodecsConfigurer.defaultCodecs().jackson2JsonDecoder(new Jackson2JsonDecoder(objectMapper, MediaType.APPLICATION_JSON));
                        }).build())
                .clientConnector(clientHttpConnector(iamReactiveClientProperties))
                .filter(oauth)
                .build()

Webclient usage:

webClient
                .get()
                .uri("https://link")
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(User.class)
                .log()

Logs:

10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onSubscribe([Fuseable] MonoFlatMap.FlatMapMain)
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onSubscribe([Fuseable] MonoPeekTerminal.MonoTerminalPeekSubscriber)
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | request(unbounded)
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | request(unbounded)
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.P.2 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:23.023Z [reactor-http-nio-2] INFO r.M.FlatMap.1 - | onContextUpdate(Context2{interface org.springframework.web.server.ServerWebExchange=org.springframework.web.server.adapter.DefaultServerWebExchange@5011f74c, interface org.springframework.security.core.context.SecurityContext=MonoFlatMap})
10:50:24.024Z [reactor-http-nio-2] ERROR r.M.FlatMap.1 - | onError(java.lang.IllegalStateException: Only one connection receive subscriber allowed.)
10:50:24.024Z [reactor-http-nio-2] ERROR r.M.FlatMap.1 -
java.lang.IllegalStateException: Only one connection receive subscriber allowed.
at reactor.netty.channel.FluxReceive.startReceiver(FluxReceive.java:277)
at reactor.netty.channel.FluxReceive.subscribe(FluxReceive.java:127)

Version

Spring version: 5.1.1.RELEASE
Reactor Netty version: 0.8.2.RELEASE
Reactor Core version: 3.2.2.RELEASE

Sample

https://github.com/hfgbarrigas/oauth2client-webflux-error-sample

[rwinch]

The problem seems that Spring Security is not handling a redirect properly. The redirect is happening when the request for a token is being submitted to http://dev.hold.co/as/oauth/token It then redirects to https. To fix it change your configuration to use https token-uri: https://dev.hold.co/as/oauth/token

I will investigate how to get a proper error message to the user. In the meantime, hopefully this helps you make some progress.

[rwinch]

The reason for the strange error message is SPR-17482. If you update to Spring Framework 5.1.3.BUILD-SNAPSHOT the error is now a 401.

HTTP/1.1 500 Internal Server Error
Cache-Control: no-cache, no-store, max-age=0, must-revalidate
Content-Length: 133
Content-Type: application/json;charset=UTF-8
Expires: 0
Pragma: no-cache
Referrer-Policy: no-referrer
X-Content-Type-Options: nosniff
X-Frame-Options: DENY
X-XSS-Protection: 1 ; mode=block

{
    "error": "Internal Server Error",
    "message": "401 Unauthorized",
    "path": "/test",
    "status": 500,
    "timestamp": "2018-11-14T02:13:55.091+0000"
}

This makes sense since the token was unable to be gotten. We should obviously improve our error message, but at least now all the pieces make sense.

[hfgbarrigas]

The error completely threw me off.. You're right, changing to https seems to work. However, when dealing with additional information in the token response (<String, Object>) jackson throws an exception. Looking at OAuth2AccessTokenResponseBody class to extract the OAuth2AccessTokenResponse:

@Override
	public Mono<OAuth2AccessTokenResponse> extract(ReactiveHttpInputMessage inputMessage,
			Context context) {
		ParameterizedTypeReference<Map<String, String>> type = new ParameterizedTypeReference<Map<String, String>>() {};
		BodyExtractor<Mono<Map<String, String>>, ReactiveHttpInputMessage> delegate = BodyExtractors.toMono(type);
		return delegate.extract(inputMessage, context)
				.map(json -> parse(json))
				.flatMap(OAuth2AccessTokenResponseBodyExtractor::oauth2AccessTokenResponse)
				.map(OAuth2AccessTokenResponseBodyExtractor::oauth2AccessTokenResponse);
	}

Looks like you're forcing Map<String, String> and down the road on parse (line 66) new JSONObject() can deal with Map<String, ?>.

Here's the error thrown:

Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize instance of java.lang.String out of START_ARRAY token
at [Source: UNKNOWN; line: -1, column: -1] (through reference chain: java.util.LinkedHashMap["client_authorities"])

Here's the token response:

{"access_token":"accessToken","token_type":"bearer","expires_in":6568,"scope":"all","grant_type":"client_credentials","organization":"HOLD","client_authorities":["INTERNAL"],"jti":"72ebfbf1-2686-450b-8dd1-525a704e6aa1"}

Should I create a new ticket?

[rwinch]

@hfgbarrigas Yes please do create a new ticket with these details. Thanks!

[hfgbarrigas]

Here: 6087

Thank you.

[rwinch]

Thanks. I'm closing this in favor of the two Spring Security issues we created and the fixed Spring Framework JIRA