Skip to content

Commit

Permalink
Guard against multiple body subscriptions
Browse files Browse the repository at this point in the history
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
  • Loading branch information
poutsma committed Jan 24, 2024
1 parent 732642d commit 186d6e3
Show file tree
Hide file tree
Showing 4 changed files with 127 additions and 106 deletions.
@@ -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;
}
}
@@ -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.
Expand All @@ -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;
Expand All @@ -26,10 +25,8 @@
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;
Expand All @@ -42,45 +39,22 @@
* @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);
}


@Override
public HttpStatus getStatusCode() {
return HttpStatus.valueOf(this.message.getHead().getCode());
}

@Override
public int getRawStatusCode() {
return this.message.getHead().getCode();
super(message.getHead().getCode(),
HttpHeaders.readOnlyHttpHeaders(new HttpComponentsHeadersAdapter(message.getHead())),
adaptCookies(context),
Flux.from(message.getBody()).map(dataBufferFactory::wrap)
);
}

@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())
Expand All @@ -93,25 +67,10 @@ public MultiValueMap<String, ResponseCookie> getCookies() {
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;
}

}
@@ -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.
Expand All @@ -21,13 +21,12 @@
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;
Expand All @@ -42,57 +41,37 @@
* @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);
}


@Override
public HttpStatus getStatusCode() {
return HttpStatus.valueOf(getRawStatusCode());
super(reactiveResponse.getStatus(),
adaptHeaders(reactiveResponse),
adaptCookies(reactiveResponse),
content);
}

@Override
public int getRawStatusCode() {
return this.reactiveResponse.getStatus();
private static HttpHeaders adaptHeaders(ReactiveResponse response) {
MultiValueMap<String, String> headers = new JettyHeadersAdapter(response.getHeaders());
return HttpHeaders.readOnlyHttpHeaders(headers);
}

@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);
}

Expand All @@ -102,15 +81,4 @@ private static String parseSameSite(String headerValue) {
return (matcher.matches() ? matcher.group(1) : null);
}


@Override
public Flux<DataBuffer> getBody() {
return this.content;
}

@Override
public HttpHeaders getHeaders() {
return this.headers;
}

}
Expand Up @@ -475,7 +475,7 @@ void retrieveJsonNull(ClientHttpConnector connector) {
.retrieve()
.bodyToMono(Map.class);

StepVerifier.create(result).verifyComplete();
StepVerifier.create(result).expectComplete().verify(Duration.ofSeconds(3));
}

@ParameterizedWebClientTest // SPR-15946
Expand Down Expand Up @@ -800,7 +800,7 @@ void statusHandlerWithErrorBodyTransformation(ClientHttpConnector connector) {
MyException error = (MyException) throwable;
assertThat(error.getMessage()).isEqualTo("foofoo");
})
.verify();
.verify(Duration.ofSeconds(3));
}

@ParameterizedWebClientTest
Expand Down Expand Up @@ -842,7 +842,7 @@ void statusHandlerSuppressedErrorSignal(ClientHttpConnector connector) {

StepVerifier.create(result)
.expectNext("Internal Server error")
.verifyComplete();
.expectComplete().verify(Duration.ofSeconds(3));

expectRequestCount(1);
expectRequest(request -> {
Expand All @@ -866,7 +866,7 @@ void statusHandlerSuppressedErrorSignalWithFlux(ClientHttpConnector connector) {

StepVerifier.create(result)
.expectNext("Internal Server error")
.verifyComplete();
.expectComplete().verify(Duration.ofSeconds(3));

expectRequestCount(1);
expectRequest(request -> {
Expand Down Expand Up @@ -1024,7 +1024,7 @@ void exchangeForEmptyBodyAsVoidEntity(ClientHttpConnector connector) {

StepVerifier.create(result)
.assertNext(r -> assertThat(r.getStatusCode().is2xxSuccessful()).isTrue())
.verifyComplete();
.expectComplete().verify(Duration.ofSeconds(3));
}

@ParameterizedWebClientTest
Expand Down Expand Up @@ -1209,7 +1209,7 @@ void malformedResponseChunksOnBodilessEntity(ClientHttpConnector connector) {
WebClientException ex = (WebClientException) throwable;
assertThat(ex.getCause()).isInstanceOf(IOException.class);
})
.verify();
.verify(Duration.ofSeconds(3));
}

@ParameterizedWebClientTest
Expand All @@ -1221,7 +1221,7 @@ void malformedResponseChunksOnEntityWithBody(ClientHttpConnector connector) {
WebClientException ex = (WebClientException) throwable;
assertThat(ex.getCause()).isInstanceOf(IOException.class);
})
.verify();
.verify(Duration.ofSeconds(3));
}

private <T> Mono<T> doMalformedChunkedResponseTest(
Expand Down

2 comments on commit 186d6e3

@sbordet
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@poutsma unfortunately this commit broke Jetty's Reactive HttpClient 2.0.x.

In Jetty Reactive HttpClient 2.0.x we depend on Jetty 10.0.x and Java 11, so we cannot depend on Spring 6.0.x because it requires Java 17.

The previous version of this class (v5.3.31) was doing some reflection based on the Jetty version that was present, and adapting Jetty's HttpFields class to a Map.
This was necessary because in 9.4.x HttpFields is a class, while in 10.0.x+ it is an interface.

With v5.3.31 there are no problems using Spring in either Jetty Reactive HttpClient 1.1.x (based on 9.4.x) or 2.0.x (based on 10.0.x).

But with Spring v5.3.32 Jetty Reactive HttpClient 1.1.x works fine, but 2.0.x now throws this:

java.lang.IncompatibleClassChangeError: Found interface org.eclipse.jetty.http.HttpFields, but class was expected
	at org.springframework.http.client.reactive.JettyClientHttpResponse.adaptCookies(JettyClientHttpResponse.java:63)
	at org.springframework.http.client.reactive.JettyClientHttpResponse.<init>(JettyClientHttpResponse.java:53)
	at org.springframework.http.client.reactive.JettyClientHttpConnector.lambda$execute$0(JettyClientHttpConnector.java:132)
	at org.eclipse.jetty.reactive.client@2.0.13-SNAPSHOT/org.eclipse.jetty.reactive.client.internal.ResponseListenerProcessor.onHeaders(ResponseListenerProcessor.java:94)
	at org.eclipse.jetty.client@10.0.20/org.eclipse.jetty.client.ResponseNotifier.notifyHeaders(ResponseNotifier.java:95)
	at org.eclipse.jetty.client@10.0.20/org.eclipse.jetty.client.ResponseNotifier.notifyHeaders(ResponseNotifier.java:87)
	at org.eclipse.jetty.client@10.0.20/org.eclipse.jetty.client.HttpReceiver.responseHeaders(HttpReceiver.java:327)
	at org.eclipse.jetty.client@10.0.20/org.eclipse.jetty.client.http.HttpReceiverOverHTTP.headerComplete(HttpReceiverOverHTTP.java:331)

Would it be possible to restore the reflection code that was removed and keep the compatibility with both 9.4.x and 10.0.x?
Let me know if I have to file an issue, or you need clarifications.
Thanks!

@poutsma
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for making us aware of this mistake, @sbordet. I have created #32337 to resolve it.

Please sign in to comment.