Fix WebClient Memory Leaks

WebClient exchange requires that the body is consumed. Before this commit
there were places where an Exception was thrown without consuming the body
if the status was not successful. There was also the potential for the
statusCode invocation to throw an Exception of the status code was not
defined which would cause a leak.

This commit ensures that before the Exception is thrown the body is
consumed. It also uses the http status in a way that will ensure an
Exception is not thrown.

Fixes gh-7293

Author: rwinch

oauth2/oauth2-client/src/main/java/org/springframework/security/oauth2/client/endpoint/WebClientReactiveClientCredentialsTokenResponseClient.java

@@ -15,7 +15,10 @@
  */
 package org.springframework.security.oauth2.client.endpoint;
 
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.security.oauth2.client.registration.ClientRegistration;
 import org.springframework.security.oauth2.core.ClientAuthenticationMethod;
@@ -66,15 +69,18 @@ public Mono<OAuth2AccessTokenResponse> getTokenResponse(OAuth2ClientCredentialsG
 					.headers(headers(clientRegistration))
 					.body(body)
 					.exchange()
-					.flatMap(response ->{
-						if (!response.statusCode().is2xxSuccessful()){
+					.flatMap(response -> {
+						HttpStatus status = HttpStatus.resolve(response.rawStatusCode());
+						if (status == null || !status.is2xxSuccessful()) {
 							// extract the contents of this into a method named oauth2AccessTokenResponse but has an argument for the response
-							throw WebClientResponseException.create(response.rawStatusCode(),
+							return response.bodyToFlux(DataBuffer.class)
+								.map(DataBufferUtils::release)
+								.then(Mono.error(WebClientResponseException.create(response.rawStatusCode(),
 											"Cannot get token, expected 2xx HTTP Status code",
 											null,
 											null,
 											null
-									);
+								)));
 						}
 						return response.body(oauth2AccessTokenResponse()); })
 					.map(response -> {

oauth2/oauth2-resource-server/src/main/java/org/springframework/security/oauth2/server/resource/introspection/NimbusReactiveOpaqueTokenIntrospector.java

@@ -28,6 +28,8 @@
 import com.nimbusds.oauth2.sdk.TokenIntrospectionSuccessResponse;
 import com.nimbusds.oauth2.sdk.http.HTTPResponse;
 import com.nimbusds.oauth2.sdk.id.Audience;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import reactor.core.publisher.Mono;
 
 import org.springframework.http.HttpHeaders;
@@ -116,8 +118,10 @@ private Mono<HTTPResponse> adaptToNimbusResponse(ClientResponse responseEntity)
 		HTTPResponse response = new HTTPResponse(responseEntity.rawStatusCode());
 		response.setHeader(HttpHeaders.CONTENT_TYPE, responseEntity.headers().contentType().get().toString());
 		if (response.getStatusCode() != HTTPResponse.SC_OK) {
-			throw new OAuth2IntrospectionException(
-					"Introspection endpoint responded with " + response.getStatusCode());
+			return responseEntity.bodyToFlux(DataBuffer.class)
+				.map(DataBufferUtils::release)
+				.then(Mono.error(new OAuth2IntrospectionException(
+					"Introspection endpoint responded with " + response.getStatusCode())));
 		}
 		return responseEntity.bodyToMono(String.class)
 				.doOnNext(response::setContent)