Add functionality to set custom web client in ReactiveOidcIdTokenDecoderFactory

[ubaid4j]

Current Behavior

On creating OIDC token (see: code) ReactiveOidcIdTokenDecoderFactory creates the NimbusReactiveJwtDecoder (see: code) that uses the default webclient (see: code). With default web client we see the following exception, see reactor/reactor-netty#1774.

2023-06-01 09:34:18,049 WARN  [reactor-http-epoll-3] HttpClientConnect: [5e0df2c6-5, L:/xx.xx.xx.xxx:xxxxx - R:auth.domain-dev.io/xx.xx.xx.xxx:443] The connection observed an error, the request cannot be retried as the headers/body were sent
io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
2023-06-01 09:34:18,053 ERROR [reactor-http-epoll-3] HttpWebHandlerAdapter: [744d1071-414] 500 Server Error for HTTP GET "/login/oauth2/code/realm?parameter==parametervalue&state=state&session_state=session_state&code=code"
java.lang.IllegalStateException: Could not obtain the keys
	at org.springframework.security.oauth2.jwt.NimbusReactiveJwtDecoder$JwkSetUriReactiveJwtDecoderBuilder.lambda$processor$6(NimbusReactiveJwtDecoder.java:398)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
	*__checkpoint ⇢ org.springframework.security.oauth2.client.web.server.authentication.OAuth2LoginAuthenticationWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.oauth2.client.web.server.OAuth2AuthorizationRequestRedirectWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ tech.jhipster.web.filter.reactive.CookieCsrfFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.web.cors.reactive.CorsWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ filter [DefaultWebFilterChain]
	*__checkpoint ⇢ filter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
	*__checkpoint ⇢ org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter [DefaultWebFilterChain]
	*__checkpoint ⇢ HTTP GET "/login/oauth2/code/realm?parameter=parameterValue&state=stateValue&session_state=sessionStateValue&code=codeValue" [ExceptionHandlingWebHandler]
Original Stack Trace:
		at org.springframework.security.oauth2.jwt.NimbusReactiveJwtDecoder$JwkSetUriReactiveJwtDecoderBuilder.lambda$processor$6(NimbusReactiveJwtDecoder.java:398)
		at reactor.core.publisher.Mono.lambda$onErrorMap$31(Mono.java:3730)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:94)
		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onError(Operators.java:2063)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onError(MonoFlatMap.java:172)
		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onError(Operators.java:2063)
		at reactor.core.publisher.Operators$MonoSubscriber.onError(Operators.java:1863)
		at reactor.core.publisher.MonoCacheTime$CoordinatorSubscriber.signalCached(MonoCacheTime.java:340)
		at reactor.core.publisher.MonoCacheTime$CoordinatorSubscriber.onError(MonoCacheTime.java:363)
		at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onError(FluxPeekFuseable.java:234)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onError(FluxMapFuseable.java:142)
		at reactor.core.publisher.MonoFlatMap$FlatMapMain.onError(MonoFlatMap.java:172)
		at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onError(Operators.java:2063)
		at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:134)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:106)
		at reactor.core.publisher.Operators.error(Operators.java:198)
		at reactor.core.publisher.MonoErrorSupplied.subscribe(MonoErrorSupplied.java:56)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4397)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.MonoNext$NextSubscriber.onError(MonoNext.java:93)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onError(MonoFlatMapMany.java:204)
		at reactor.core.publisher.SerializedSubscriber.onError(SerializedSubscriber.java:124)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.whenError(FluxRetryWhen.java:225)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onError(FluxRetryWhen.java:274)
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:415)
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:251)
		at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:537)
		at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:343)
		at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100)
		at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:190)
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:201)
		at reactor.netty.http.client.HttpClientConnect$HttpObserver.onUncaughtException(HttpClientConnect.java:400)
		at reactor.netty.ReactorNetty$CompositeConnectionObserver.onUncaughtException(ReactorNetty.java:670)
		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onUncaughtException(DefaultPooledConnectionProvider.java:205)
		at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onUncaughtException(DefaultPooledConnectionProvider.java:454)
		at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:232)
		at reactor.netty.channel.FluxReceive.onInboundError(FluxReceive.java:453)
		at reactor.netty.channel.ChannelOperations.onInboundError(ChannelOperations.java:488)
		at reactor.netty.channel.ChannelOperationsHandler.exceptionCaught(ChannelOperationsHandler.java:126)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireExceptionCaught(CombinedChannelDuplexHandler.java:424)
		at io.netty.channel.ChannelHandlerAdapter.exceptionCaught(ChannelHandlerAdapter.java:92)
		at io.netty.channel.CombinedChannelDuplexHandler$1.fireExceptionCaught(CombinedChannelDuplexHandler.java:145)
		at io.netty.channel.ChannelInboundHandlerAdapter.exceptionCaught(ChannelInboundHandlerAdapter.java:143)
		at io.netty.channel.CombinedChannelDuplexHandler.exceptionCaught(CombinedChannelDuplexHandler.java:231)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273)
		at io.netty.handler.ssl.SslHandler.exceptionCaught(SslHandler.java:1105)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.exceptionCaught(DefaultChannelPipeline.java:1377)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.DefaultChannelPipeline.fireExceptionCaught(DefaultChannelPipeline.java:907)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.handleReadException(AbstractEpollStreamChannel.java:728)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:826)
		at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:487)
		at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:385)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
		at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.springframework.web.reactive.function.client.WebClientRequestException: readAddress(..) failed: Connection reset by peer; nested exception is io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer
	at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:141)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Error has been observed at the following site(s):
	*__checkpoint ⇢ Request to GET https://auth-server/auth/realms/realmName/protocol/openid-connect/certs [DefaultWebClient]
Original Stack Trace:
		at org.springframework.web.reactive.function.client.ExchangeFunctions$DefaultExchangeFunction.lambda$wrapException$9(ExchangeFunctions.java:141)
		at reactor.core.publisher.MonoErrorSupplied.subscribe(MonoErrorSupplied.java:55)
		at reactor.core.publisher.Mono.subscribe(Mono.java:4397)
		at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:103)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222)
		at reactor.core.publisher.MonoNext$NextSubscriber.onError(MonoNext.java:93)
		at reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onError(MonoFlatMapMany.java:204)
		at reactor.core.publisher.SerializedSubscriber.onError(SerializedSubscriber.java:124)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.whenError(FluxRetryWhen.java:225)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onError(FluxRetryWhen.java:274)
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:415)
		at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onNext(FluxConcatMap.java:251)
		at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:537)
		at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:343)
		at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100)
		at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
		at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.onError(FluxRetryWhen.java:190)
		at reactor.core.publisher.MonoCreate$DefaultMonoSink.error(MonoCreate.java:201)
		at reactor.netty.http.client.HttpClientConnect$HttpObserver.onUncaughtException(HttpClientConnect.java:400)
		at reactor.netty.ReactorNetty$CompositeConnectionObserver.onUncaughtException(ReactorNetty.java:670)
		at reactor.netty.resources.DefaultPooledConnectionProvider$DisposableAcquire.onUncaughtException(DefaultPooledConnectionProvider.java:205)
		at reactor.netty.resources.DefaultPooledConnectionProvider$PooledConnection.onUncaughtException(DefaultPooledConnectionProvider.java:454)
		at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:232)
		at reactor.netty.channel.FluxReceive.onInboundError(FluxReceive.java:453)
		at reactor.netty.channel.ChannelOperations.onInboundError(ChannelOperations.java:488)
		at reactor.netty.channel.ChannelOperationsHandler.exceptionCaught(ChannelOperationsHandler.java:126)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273)
		at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireExceptionCaught(CombinedChannelDuplexHandler.java:424)
		at io.netty.channel.ChannelHandlerAdapter.exceptionCaught(ChannelHandlerAdapter.java:92)
		at io.netty.channel.CombinedChannelDuplexHandler$1.fireExceptionCaught(CombinedChannelDuplexHandler.java:145)
		at io.netty.channel.ChannelInboundHandlerAdapter.exceptionCaught(ChannelInboundHandlerAdapter.java:143)
		at io.netty.channel.CombinedChannelDuplexHandler.exceptionCaught(CombinedChannelDuplexHandler.java:231)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273)
		at io.netty.handler.ssl.SslHandler.exceptionCaught(SslHandler.java:1105)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.AbstractChannelHandlerContext.fireExceptionCaught(AbstractChannelHandlerContext.java:273)
		at io.netty.channel.DefaultChannelPipeline$HeadContext.exceptionCaught(DefaultChannelPipeline.java:1377)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:302)
		at io.netty.channel.AbstractChannelHandlerContext.invokeExceptionCaught(AbstractChannelHandlerContext.java:281)
		at io.netty.channel.DefaultChannelPipeline.fireExceptionCaught(DefaultChannelPipeline.java:907)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.handleReadException(AbstractEpollStreamChannel.java:728)
		at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:826)
		at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:487)
		at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:385)
		at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
		at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
		at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
		at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer

Expected behavior

To fix the above issue, we can create our custom web client using a custom connection provider. see: reactor/reactor-netty#1774 (comment), and then pass this web client to ReactiveOidcIdTokenDecoderFactory that creates the NimbusReactiveJwtDecoder with custom web client thus we can fix above exception.
The expectation is that ReactiveOidcIdTokenDecoderFactory should have the functionality to set a custom web client and creates NimbusReactiveJwtDecoder with that custom web client.

Following is the required change to get this behavior:

--- a/oauth2/oauth2-client/src/main/java/org/springframework/security/oauth2/client/oidc/authentication/ReactiveOidcIdTokenDecoderFactory.java	(revision 25e6c51a0b3f8c1272e85509391847e1da8f1d16)
+++ b/oauth2/oauth2-client/src/main/java/org/springframework/security/oauth2/client/oidc/authentication/ReactiveOidcIdTokenDecoderFactory.java	(revision 4ca3452dc4019cfef01869ce2c015bd870a71e71)
@@ -49,6 +49,7 @@
 import org.springframework.security.oauth2.jwt.ReactiveJwtDecoderFactory;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
+import org.springframework.web.reactive.function.client.WebClient;
 
 /**
  * A {@link ReactiveJwtDecoderFactory factory} that provides a {@link ReactiveJwtDecoder}
@@ -89,6 +90,7 @@
 	private Function<ClientRegistration, Converter<Map<String, Object>, Map<String, Object>>> claimTypeConverterFactory = (
 			clientRegistration) -> DEFAULT_CLAIM_TYPE_CONVERTER;
 
+	private WebClient webClient = WebClient.create();
 	/**
 	 * Returns the default {@link Converter}'s used for type conversion of claim values
 	 * for an {@link OidcIdToken}.
@@ -164,6 +166,7 @@
 				throw new OAuth2AuthenticationException(oauth2Error, oauth2Error.toString());
 			}
 			return NimbusReactiveJwtDecoder.withJwkSetUri(jwkSetUri).jwsAlgorithm((SignatureAlgorithm) jwsAlgorithm)
+					.webClient(webClient)
 					.build();
 		}
 		if (jwsAlgorithm != null && MacAlgorithm.class.isAssignableFrom(jwsAlgorithm.getClass())) {
@@ -239,4 +242,15 @@
 		this.claimTypeConverterFactory = claimTypeConverterFactory;
 	}
 
+	/**
+	 * Sets the custom web client that will be used in {@link NimbusReactiveJwtDecoder}.
+	 * The default webClient is created by {@code WebClient.create()}.
+	 * This is optional method if we need to set custom web client in {@link NimbusReactiveJwtDecoder}.
+	 *
+	 * @param webClient webclient
+	 */
+	public void setWebClient(WebClient webClient) {
+		this.webClient = webClient;
+	}
+
 }

and we can use the following code to set the custom web client in ReactiveOidcIdTokenDecoderFactory in our applications.

    @Bean
    ReactiveJwtDecoderFactory<ClientRegistration> reactiveJwtDecoderFactory(WebClient webClient) {
        ReactiveOidcIdTokenDecoderFactory reactiveJwtDecoderFactory = new ReactiveOidcIdTokenDecoderFactory();
        reactiveJwtDecoderFactory.setWebClient(webClient); // this is new expected functionality 
        return reactiveJwtDecoderFactory;
    }

Context
After migrating from Zuul API gateway to Spring Cloud gateway we rewrote our whole multi-tenant oidc login flow in spring-security-oauth2-client (migrated from Keycloak adapter).
After deployment in dev env, we started getting above listed exception in four places.

1. reactive fiegn clients (fixed by providing custom http client)
2. reactor netty routes (fixed by providing custom tcp client)
3. In AbstractWebClientReactiveOAuth2AccessTokenResponseClient#getTokenResponse (fixed by providing custom web client)
   see

    @Bean
    AbstractWebClientReactiveOAuth2AccessTokenResponseClient<OAuth2AuthorizationCodeGrantRequest> accessTokenResponseClient(WebClient webClient) {
        AbstractWebClientReactiveOAuth2AccessTokenResponseClient<OAuth2AuthorizationCodeGrantRequest> accessTokenResponseClient
            = new WebClientReactiveAuthorizationCodeTokenResponseClient();
        accessTokenResponseClient.setWebClient(webClient);
        return accessTokenResponseClient;
    }

4. In org.springframework.security.oauth2.jwt.NimbusReactiveJwtDecoder$JwkSetUriReactiveJwtDecoderBuilder.lambda$processor$6(NimbusReactiveJwtDecoder.java:398) (This issue is created to fix exception here).

So, to fix issue#4, we need the same functionality (to set web client) as for AbstractWebClientReactiveOAuth2AccessTokenResponseClient. see issue#3.

There are workarounds like implementing ReactiveJwtDecoderFactory<ClientRegistration> as per our need but I am not considering it as we have already an implementation of ReactiveJwtDecoderFactory<ClientRegistration> and we just need to set a custom web client.

Let me know if there is a need for more information. I am happy to create PR for this enhancement.

Thanks