Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Guard against multiple body subscriptions

Browse Source 
Before this commit, the Jetty connector did not have any
safeguard against multiple body subscriptions. Such as check has now
been added.

See gh-32100
Closes gh-32101
This commit is contained in:
Arjen Poutsma
2024-01-24 13:24:55 +01:00
parent 732642d8ad
commit 186d6e370d
4 changed files with 127 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
spring-web/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpResponse.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* Copyright 2002-2024 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.http.client.reactive;
import java.util.concurrent.atomic.AtomicBoolean;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseCookie;
import org.springframework.util.Assert;
import org.springframework.util.MultiValueMap;
/**
* Base class for {@link ClientHttpResponse} implementations.
*
* @author Arjen Poutsma
* @since 5.3.32
*/
public abstract class AbstractClientHttpResponse implements ClientHttpResponse {
private final int statusCode;
private final HttpHeaders headers;
private final MultiValueMap<String, ResponseCookie> cookies;
private final Flux<DataBuffer> body;
protected AbstractClientHttpResponse(int statusCode, HttpHeaders headers,
MultiValueMap<String, ResponseCookie> cookies, Flux<DataBuffer> body) {
Assert.notNull(headers, "Headers must not be null");
Assert.notNull(body, "Body must not be null");
this.statusCode = statusCode;
this.headers = headers;
this.cookies = cookies;
this.body = singleSubscription(body);
}
private static Flux<DataBuffer> singleSubscription(Flux<DataBuffer> body) {
AtomicBoolean subscribed = new AtomicBoolean();
return body.doOnSubscribe(s -> {
if (!subscribed.compareAndSet(false, true)) {
throw new IllegalStateException("The client response body can only be consumed once");
}
});
}
@Override
public HttpStatus getStatusCode() {
return HttpStatus.valueOf(this.statusCode);
}
@Override
public int getRawStatusCode() {
return this.statusCode;
}
@Override
public HttpHeaders getHeaders() {
return this.headers;
}
@Override
public MultiValueMap<String, ResponseCookie> getCookies() {
return this.cookies;
}
@Override
public Flux<DataBuffer> getBody() {
return this.body;
}
}

61
spring-web/src/main/java/org/springframework/http/client/reactive/HttpComponentsClientHttpResponse.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2024 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.
@@ -17,7 +17,6 @@
package org.springframework.http.client.reactive;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hc.client5.http.cookie.Cookie;
import org.apache.hc.client5.http.protocol.HttpClientContext;
@@ -26,10 +25,8 @@ import org.apache.hc.core5.http.Message;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseCookie;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
@@ -42,45 +39,22 @@ import org.springframework.util.MultiValueMap;
* @since 5.3
* @see <a href="https://hc.apache.org/index.html">Apache HttpComponents</a>
*/
class HttpComponentsClientHttpResponse implements ClientHttpResponse {
private final DataBufferFactory dataBufferFactory;
private final Message<HttpResponse, Publisher<ByteBuffer>> message;
private final HttpHeaders headers;
private final HttpClientContext context;
private final AtomicBoolean rejectSubscribers = new AtomicBoolean();
class HttpComponentsClientHttpResponse extends AbstractClientHttpResponse {
public HttpComponentsClientHttpResponse(DataBufferFactory dataBufferFactory,
Message<HttpResponse, Publisher<ByteBuffer>> message, HttpClientContext context) {
this.dataBufferFactory = dataBufferFactory;
this.message = message;
this.context = context;
MultiValueMap<String, String> adapter = new HttpComponentsHeadersAdapter(message.getHead());
this.headers = HttpHeaders.readOnlyHttpHeaders(adapter);
super(message.getHead().getCode(),
HttpHeaders.readOnlyHttpHeaders(new HttpComponentsHeadersAdapter(message.getHead())),
adaptCookies(context),
Flux.from(message.getBody()).map(dataBufferFactory::wrap)
);
}
@Override
public HttpStatus getStatusCode() {
return HttpStatus.valueOf(this.message.getHead().getCode());
}
@Override
public int getRawStatusCode() {
return this.message.getHead().getCode();
}
@Override
public MultiValueMap<String, ResponseCookie> getCookies() {
private static MultiValueMap<String, ResponseCookie> adaptCookies(HttpClientContext context) {
LinkedMultiValueMap<String, ResponseCookie> result = new LinkedMultiValueMap<>();
this.context.getCookieStore().getCookies().forEach(cookie ->
context.getCookieStore().getCookies().forEach(cookie ->
result.add(cookie.getName(),
ResponseCookie.fromClientResponse(cookie.getName(), cookie.getValue())
.domain(cookie.getDomain())
@@ -93,25 +67,10 @@ class HttpComponentsClientHttpResponse implements ClientHttpResponse {
return result;
}
private long getMaxAgeSeconds(Cookie cookie) {
private static long getMaxAgeSeconds(Cookie cookie) {
String maxAgeAttribute = cookie.getAttribute(Cookie.MAX_AGE_ATTR);
return (maxAgeAttribute != null ? Long.parseLong(maxAgeAttribute) : -1);
}
@Override
public Flux<DataBuffer> getBody() {
return Flux.from(this.message.getBody())
.doOnSubscribe(s -> {
if (!this.rejectSubscribers.compareAndSet(false, true)) {
throw new IllegalStateException("The client response body can only be consumed once.");
}
})
.map(this.dataBufferFactory::wrap);
}
@Override
public HttpHeaders getHeaders() {
return this.headers;
}
}

64
spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpResponse.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2024 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.
@@ -21,13 +21,12 @@ import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.eclipse.jetty.http.HttpField;
import org.eclipse.jetty.reactive.client.ReactiveResponse;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseCookie;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;
@@ -42,57 +41,37 @@ import org.springframework.util.MultiValueMap;
* @see <a href="https://github.com/jetty-project/jetty-reactive-httpclient">
* Jetty ReactiveStreams HttpClient</a>
*/
class JettyClientHttpResponse implements ClientHttpResponse {
class JettyClientHttpResponse extends AbstractClientHttpResponse {
private static final Pattern SAMESITE_PATTERN = Pattern.compile("(?i).*SameSite=(Strict|Lax|None).*");
private final ReactiveResponse reactiveResponse;
public JettyClientHttpResponse(ReactiveResponse reactiveResponse, Flux<DataBuffer> content) {
private final Flux<DataBuffer> content;
private final HttpHeaders headers;
public JettyClientHttpResponse(ReactiveResponse reactiveResponse, Publisher<DataBuffer> content) {
this.reactiveResponse = reactiveResponse;
this.content = Flux.from(content);
MultiValueMap<String, String> headers = (Jetty10HttpFieldsHelper.jetty10Present() ?
Jetty10HttpFieldsHelper.getHttpHeaders(reactiveResponse.getResponse()) :
new JettyHeadersAdapter(reactiveResponse.getHeaders()));
this.headers = HttpHeaders.readOnlyHttpHeaders(headers);
super(reactiveResponse.getStatus(),
adaptHeaders(reactiveResponse),
adaptCookies(reactiveResponse),
content);
}
@Override
public HttpStatus getStatusCode() {
return HttpStatus.valueOf(getRawStatusCode());
private static HttpHeaders adaptHeaders(ReactiveResponse response) {
MultiValueMap<String, String> headers = new JettyHeadersAdapter(response.getHeaders());
return HttpHeaders.readOnlyHttpHeaders(headers);
}
@Override
public int getRawStatusCode() {
return this.reactiveResponse.getStatus();
}
@Override
public MultiValueMap<String, ResponseCookie> getCookies() {
private static MultiValueMap<String, ResponseCookie> adaptCookies(ReactiveResponse response) {
MultiValueMap<String, ResponseCookie> result = new LinkedMultiValueMap<>();
List<String> cookieHeader = getHeaders().get(HttpHeaders.SET_COOKIE);
if (cookieHeader != null) {
cookieHeader.forEach(header ->
HttpCookie.parse(header).forEach(cookie -> result.add(cookie.getName(),
List<HttpField> cookieHeaders = response.getHeaders().getFields(HttpHeaders.SET_COOKIE);
cookieHeaders.forEach(header ->
HttpCookie.parse(header.getValue()).forEach(cookie -> result.add(cookie.getName(),
ResponseCookie.fromClientResponse(cookie.getName(), cookie.getValue())
.domain(cookie.getDomain())
.path(cookie.getPath())
.maxAge(cookie.getMaxAge())
.secure(cookie.getSecure())
.httpOnly(cookie.isHttpOnly())
.sameSite(parseSameSite(header))
.sameSite(parseSameSite(header.getValue()))
.build()))
);
}
return CollectionUtils.unmodifiableMultiValueMap(result);
}
@@ -102,15 +81,4 @@ class JettyClientHttpResponse implements ClientHttpResponse {
return (matcher.matches() ? matcher.group(1) : null);
}
@Override
public Flux<DataBuffer> getBody() {
return this.content;
}
@Override
public HttpHeaders getHeaders() {
return this.headers;
}
}

14
spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java
View File

@@ -475,7 +475,7 @@ class WebClientIntegrationTests {
.retrieve()
.bodyToMono(Map.class);
StepVerifier.create(result).verifyComplete();
StepVerifier.create(result).expectComplete().verify(Duration.ofSeconds(3));
}
@ParameterizedWebClientTest // SPR-15946
@@ -800,7 +800,7 @@ class WebClientIntegrationTests {
MyException error = (MyException) throwable;
assertThat(error.getMessage()).isEqualTo("foofoo");
})
.verify();
.verify(Duration.ofSeconds(3));
}
@ParameterizedWebClientTest
@@ -842,7 +842,7 @@ class WebClientIntegrationTests {
StepVerifier.create(result)
.expectNext("Internal Server error")
.verifyComplete();
.expectComplete().verify(Duration.ofSeconds(3));
expectRequestCount(1);
expectRequest(request -> {
@@ -866,7 +866,7 @@ class WebClientIntegrationTests {
StepVerifier.create(result)
.expectNext("Internal Server error")
.verifyComplete();
.expectComplete().verify(Duration.ofSeconds(3));
expectRequestCount(1);
expectRequest(request -> {
@@ -1024,7 +1024,7 @@ class WebClientIntegrationTests {
StepVerifier.create(result)
.assertNext(r -> assertThat(r.getStatusCode().is2xxSuccessful()).isTrue())
.verifyComplete();
.expectComplete().verify(Duration.ofSeconds(3));
}
@ParameterizedWebClientTest
@@ -1209,7 +1209,7 @@ class WebClientIntegrationTests {
WebClientException ex = (WebClientException) throwable;
assertThat(ex.getCause()).isInstanceOf(IOException.class);
})
.verify();
.verify(Duration.ofSeconds(3));
}
@ParameterizedWebClientTest
@@ -1221,7 +1221,7 @@ class WebClientIntegrationTests {
WebClientException ex = (WebClientException) throwable;
assertThat(ex.getCause()).isInstanceOf(IOException.class);
})
.verify();
.verify(Duration.ofSeconds(3));
}
private <T> Mono<T> doMalformedChunkedResponseTest(