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 JDK and Jetty connectors do not have any
safeguards against multiple body subscriptions. Such as check has now
been added.

See gh-32100
Closes gh-32102
  • Loading branch information
@poutsma
poutsma committed Jan 24, 2024
1 parent 3817936 commit c749a14
Show file tree
Hide file tree
Showing 5 changed files with 140 additions and 124 deletions.
90 changes: 90 additions & 0 deletions ...eb/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpResponse.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
/*
* 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.HttpStatusCode;
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 HttpStatusCode statusCode;

private final HttpHeaders headers;

private final MultiValueMap<String, ResponseCookie> cookies;

private final Flux<DataBuffer> body;



protected AbstractClientHttpResponse(HttpStatusCode statusCode, HttpHeaders headers,
MultiValueMap<String, ResponseCookie> cookies, Flux<DataBuffer> body) {

Assert.notNull(statusCode, "StatusCode must not be null");
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 HttpStatusCode getStatusCode() {
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;
}
}
55 changes: 10 additions & 45 deletions .../main/java/org/springframework/http/client/reactive/HttpComponentsClientHttpResponse.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 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,7 +25,6 @@
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.HttpStatusCode;
Expand All @@ -42,40 +40,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);
super(HttpStatusCode.valueOf(message.getHead().getCode()),
HttpHeaders.readOnlyHttpHeaders(new HttpComponentsHeadersAdapter(message.getHead())),
adaptCookies(context),
Flux.from(message.getBody()).map(dataBufferFactory::wrap)
);
}


@Override
public HttpStatusCode getStatusCode() {
return HttpStatusCode.valueOf(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())
Expand All @@ -88,25 +68,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;
}

}
50 changes: 17 additions & 33 deletions spring-web/src/main/java/org/springframework/http/client/reactive/JdkClientHttpResponse.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 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 Down Expand Up @@ -50,24 +50,20 @@
* @author Rossen Stoyanchev
* @since 6.0
*/
class JdkClientHttpResponse implements ClientHttpResponse {
class JdkClientHttpResponse extends AbstractClientHttpResponse {

private static final Pattern SAME_SITE_PATTERN = Pattern.compile("(?i).*SameSite=(Strict|Lax|None).*");


private final HttpResponse<Flow.Publisher<List<ByteBuffer>>> response;

private final DataBufferFactory bufferFactory;
public JdkClientHttpResponse(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response,
DataBufferFactory bufferFactory) {

private final HttpHeaders headers;


public JdkClientHttpResponse(
HttpResponse<Flow.Publisher<List<ByteBuffer>>> response, DataBufferFactory bufferFactory) {

this.response = response;
this.bufferFactory = bufferFactory;
this.headers = adaptHeaders(response);
super(HttpStatusCode.valueOf(response.statusCode()),
adaptHeaders(response),
adaptCookies(response),
adaptBody(response, bufferFactory)
);
}

private static HttpHeaders adaptHeaders(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
Expand All @@ -78,20 +74,8 @@ private static HttpHeaders adaptHeaders(HttpResponse<Flow.Publisher<List<ByteBuf
return HttpHeaders.readOnlyHttpHeaders(multiValueMap);
}


@Override
public HttpStatusCode getStatusCode() {
return HttpStatusCode.valueOf(this.response.statusCode());
}

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

@Override
public MultiValueMap<String, ResponseCookie> getCookies() {
return this.response.headers().allValues(HttpHeaders.SET_COOKIE).stream()
private static MultiValueMap<String, ResponseCookie> adaptCookies(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
return response.headers().allValues(HttpHeaders.SET_COOKIE).stream()
.flatMap(header -> {
Matcher matcher = SAME_SITE_PATTERN.matcher(header);
String sameSite = (matcher.matches() ? matcher.group(1) : null);
Expand All @@ -102,7 +86,7 @@ public MultiValueMap<String, ResponseCookie> getCookies() {
LinkedMultiValueMap::addAll);
}

private ResponseCookie toResponseCookie(HttpCookie cookie, @Nullable String sameSite) {
private static ResponseCookie toResponseCookie(HttpCookie cookie, @Nullable String sameSite) {
return ResponseCookie.from(cookie.getName(), cookie.getValue())
.domain(cookie.getDomain())
.httpOnly(cookie.isHttpOnly())
Expand All @@ -113,12 +97,12 @@ private ResponseCookie toResponseCookie(HttpCookie cookie, @Nullable String same
.build();
}

@Override
public Flux<DataBuffer> getBody() {
return JdkFlowAdapter.flowPublisherToFlux(this.response.body())
private static Flux<DataBuffer> adaptBody(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response, DataBufferFactory bufferFactory) {
return JdkFlowAdapter.flowPublisherToFlux(response.body())
.flatMapIterable(Function.identity())
.map(this.bufferFactory::wrap)
.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
.map(bufferFactory::wrap)
.doOnDiscard(DataBuffer.class, DataBufferUtils::release)
.cache(0);
}

}
55 changes: 16 additions & 39 deletions ...g-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpResponse.java
View file
Original file line number Diff line number Diff line change
@@ -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,8 +21,8 @@
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;
Expand All @@ -42,49 +42,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 SAME_SITE_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 = new JettyHeadersAdapter(reactiveResponse.getHeaders());
this.headers = HttpHeaders.readOnlyHttpHeaders(headers);
super(HttpStatusCode.valueOf(reactiveResponse.getStatus()),
adaptHeaders(reactiveResponse),
adaptCookies(reactiveResponse),
content);
}


@Override
public HttpStatusCode getStatusCode() {
return HttpStatusCode.valueOf(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 @@ -94,15 +82,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;
}

}

0 comments on commit c749a14

Please sign in to comment.