Limits on input stream in codecs

- Add maxInMemorySize property to Decoder and HttpMessageReader
  implementations that aggregate input to trigger
  DataBufferLimitException when reached.

- For codecs that call DataBufferUtils#join, there is now an overloaded
  variant with a maxInMemorySize extra argument. Internally, a custom
  LimitedDataBufferList is used to count and enforce the limit.

- Jackson2Tokenizer and XmlEventDecoder support those limits per
  streamed JSON object.

- Configurable limits for multipart requests with Synchronoss NIO.

- Centralized maxInMemorySize exposed via CodecConfigurer along with
  ability to plug in an instance of MultipartHttpMessageWrite.

Closes gh-23884

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2TokenizerTests.java

@@ -20,7 +20,6 @@ import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
-import java.util.function.Consumer;
 
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.TreeNode;
@@ -36,6 +35,7 @@ import reactor.test.StepVerifier;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 
 import static java.util.Arrays.*;
 import static java.util.Collections.*;
@@ -181,11 +181,68 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 		testTokenize(asList("[1", ",2,", "3]"), asList("1", "2", "3"), true);
 	}
 
+	private void testTokenize(List<String> input, List<String> output, boolean tokenize) {
+		StepVerifier.FirstStep<String> builder = StepVerifier.create(decode(input, tokenize, -1));
+		output.forEach(expected -> builder.assertNext(actual -> {
+			try {
+				JSONAssert.assertEquals(expected, actual, true);
+			}
+			catch (JSONException ex) {
+				throw new RuntimeException(ex);
+			}
+		}));
+		builder.verifyComplete();
+	}
+
+	@Test
+	public void testLimit() {
+
+		List<String> source = asList("[",
+				"{", "\"id\":1,\"name\":\"Dan\"", "},",
+				"{", "\"id\":2,\"name\":\"Ron\"", "},",
+				"{", "\"id\":3,\"name\":\"Bartholomew\"", "}",
+				"]");
+
+		String expected = String.join("", source);
+		int maxInMemorySize = expected.length();
+
+		StepVerifier.create(decode(source, false, maxInMemorySize))
+				.expectNext(expected)
+				.verifyComplete();
+
+		StepVerifier.create(decode(source, false, maxInMemorySize - 1))
+				.expectError(DataBufferLimitException.class);
+	}
+
+	@Test
+	public void testLimitTokenized() {
+
+		List<String> source = asList("[",
+				"{", "\"id\":1, \"name\":\"Dan\"", "},",
+				"{", "\"id\":2, \"name\":\"Ron\"", "},",
+				"{", "\"id\":3, \"name\":\"Bartholomew\"", "}",
+				"]");
+
+		String expected = "{\"id\":3,\"name\":\"Bartholomew\"}";
+		int maxInMemorySize = expected.length();
+
+		StepVerifier.create(decode(source, true, maxInMemorySize))
+				.expectNext("{\"id\":1,\"name\":\"Dan\"}")
+				.expectNext("{\"id\":2,\"name\":\"Ron\"}")
+				.expectNext(expected)
+				.verifyComplete();
+
+		StepVerifier.create(decode(source, true, maxInMemorySize - 1))
+				.expectNext("{\"id\":1,\"name\":\"Dan\"}")
+				.expectNext("{\"id\":2,\"name\":\"Ron\"}")
+				.verifyError(DataBufferLimitException.class);
+	}
+
 	@Test
 	public void errorInStream() {
 		DataBuffer buffer = stringBuffer("{\"id\":1,\"name\":");
 		Flux<DataBuffer> source = Flux.just(buffer).concatWith(Flux.error(new RuntimeException()));
-		Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, true);
+		Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, true, -1);
 
 		StepVerifier.create(result)
 				.expectError(RuntimeException.class)
@@ -195,7 +252,7 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 	@Test  // SPR-16521
 	public void jsonEOFExceptionIsWrappedAsDecodingError() {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"status\": \"noClosingQuote}"));
-		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, false);
+		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, false, -1);
 
 		StepVerifier.create(tokens)
 				.expectError(DecodingException.class)
@@ -203,12 +260,13 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 	}
 
 
-	private void testTokenize(List<String> source, List<String> expected, boolean tokenizeArrayElements) {
+	private Flux<String> decode(List<String> source, boolean tokenize, int maxInMemorySize) {
+
 		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
 				Flux.fromIterable(source).map(this::stringBuffer),
-				this.jsonFactory, this.objectMapper, tokenizeArrayElements);
+				this.jsonFactory, this.objectMapper, tokenize, maxInMemorySize);
 
-		Flux<String> result = tokens
+		return tokens
 				.map(tokenBuffer -> {
 					try {
 						TreeNode root = this.objectMapper.readTree(tokenBuffer.asParser());
@@ -218,10 +276,6 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 						throw new UncheckedIOException(ex);
 					}
 				});
-
-		StepVerifier.FirstStep<String> builder = StepVerifier.create(result);
-		expected.forEach(s -> builder.assertNext(new JSONAssertConsumer(s)));
-		builder.verifyComplete();
 	}
 
 	private DataBuffer stringBuffer(String value) {
@@ -231,24 +285,4 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 		return buffer;
 	}
 
-
-	private static class JSONAssertConsumer implements Consumer<String> {
-
-		private final String expected;
-
-		JSONAssertConsumer(String expected) {
-			this.expected = expected;
-		}
-
-		@Override
-		public void accept(String s) {
-			try {
-				JSONAssert.assertEquals(this.expected, s, true);
-			}
-			catch (JSONException ex) {
-				throw new RuntimeException(ex);
-			}
-		}
-	}
-
 }

spring-web/src/test/java/org/springframework/http/codec/multipart/SynchronossPartHttpMessageReaderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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,15 +17,20 @@
 package org.springframework.http.codec.multipart;
 
 import java.io.File;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import org.junit.Test;
+import org.reactivestreams.Subscription;
+import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
@@ -38,23 +43,31 @@ import org.springframework.mock.http.client.reactive.test.MockClientHttpRequest;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.util.MultiValueMap;
 
-import static java.util.Collections.*;
-import static org.junit.Assert.*;
-import static org.springframework.core.ResolvableType.*;
-import static org.springframework.http.HttpHeaders.*;
-import static org.springframework.http.MediaType.*;
+import static java.util.Collections.emptyMap;
+import static org.hamcrest.core.StringStartsWith.startsWith;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.core.ResolvableType.forClassWithGenerics;
+import static org.springframework.http.HttpHeaders.CONTENT_TYPE;
+import static org.springframework.http.MediaType.MULTIPART_FORM_DATA;
 
 /**
  * Unit tests for {@link SynchronossPartHttpMessageReader}.
  *
  * @author Sebastien Deleuze
  * @author Rossen Stoyanchev
+ * @author Brian Clozel
  */
 public class SynchronossPartHttpMessageReaderTests {
 
 	private final MultipartHttpMessageReader reader =
 			new MultipartHttpMessageReader(new SynchronossPartHttpMessageReader());
 
+	private static final ResolvableType PARTS_ELEMENT_TYPE =
+			forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
 
 	@Test
 	public void canRead() {
@@ -86,10 +99,10 @@ public class SynchronossPartHttpMessageReaderTests {
 		MultiValueMap<String, Part> parts = this.reader.readMono(elementType, request, emptyMap()).block();
 		assertEquals(2, parts.size());
 
-		assertTrue(parts.containsKey("fooPart"));
-		Part part = parts.getFirst("fooPart");
+		assertTrue(parts.containsKey("filePart"));
+		Part part = parts.getFirst("filePart");
 		assertTrue(part instanceof FilePart);
-		assertEquals("fooPart", part.name());
+		assertEquals("filePart", part.name());
 		assertEquals("foo.txt", ((FilePart) part).filename());
 		DataBuffer buffer = DataBufferUtils.join(part.content()).block();
 		assertEquals(12, buffer.readableByteCount());
@@ -97,24 +110,23 @@ public class SynchronossPartHttpMessageReaderTests {
 		buffer.read(byteContent);
 		assertEquals("Lorem Ipsum.", new String(byteContent));
 
-		assertTrue(parts.containsKey("barPart"));
-		part = parts.getFirst("barPart");
+		assertTrue(parts.containsKey("textPart"));
+		part = parts.getFirst("textPart");
 		assertTrue(part instanceof FormFieldPart);
-		assertEquals("barPart", part.name());
-		assertEquals("bar", ((FormFieldPart) part).value());
+		assertEquals("textPart", part.name());
+		assertEquals("sample-text", ((FormFieldPart) part).value());
 	}
 
 	@Test // SPR-16545
-	public void transferTo() {
+	public void transferTo() throws IOException {
 		ServerHttpRequest request = generateMultipartRequest();
-		ResolvableType elementType = forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
-		MultiValueMap<String, Part> parts = this.reader.readMono(elementType, request, emptyMap()).block();
+		MultiValueMap<String, Part> parts = this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap()).block();
 
 		assertNotNull(parts);
-		FilePart part = (FilePart) parts.getFirst("fooPart");
+		FilePart part = (FilePart) parts.getFirst("filePart");
 		assertNotNull(part);
 
-		File dest = new File(System.getProperty("java.io.tmpdir") + "/" + part.filename());
+		File dest = File.createTempFile(part.filename(), "multipart");
 		part.transferTo(dest).block(Duration.ofSeconds(5));
 
 		assertTrue(dest.exists());
@@ -125,22 +137,65 @@ public class SynchronossPartHttpMessageReaderTests {
 	@Test
 	public void bodyError() {
 		ServerHttpRequest request = generateErrorMultipartRequest();
-		ResolvableType elementType = forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
-		StepVerifier.create(this.reader.readMono(elementType, request, emptyMap())).verifyError();
+		StepVerifier.create(this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap())).verifyError();
 	}
 
+	@Test
+	public void readPartsWithoutDemand() {
+		ServerHttpRequest request = generateMultipartRequest();
+		Mono<MultiValueMap<String, Part>> parts = this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap());
+		ZeroDemandSubscriber subscriber = new ZeroDemandSubscriber();
+		parts.subscribe(subscriber);
+		subscriber.cancel();
+	}
+
+	@Test
+	public void readTooManyParts() {
+		testMultipartExceptions(reader -> reader.setMaxParts(1), ex -> {
+			assertEquals(DecodingException.class, ex.getClass());
+			assertThat(ex.getMessage(), startsWith("Failure while parsing part[2]"));
+			assertEquals("Too many parts (2 allowed)", ex.getCause().getMessage());
+		});
+	}
+
+	@Test
+	public void readFilePartTooBig() {
+		testMultipartExceptions(reader -> reader.setMaxDiskUsagePerPart(5), ex -> {
+			assertEquals(DecodingException.class, ex.getClass());
+			assertThat(ex.getMessage(), startsWith("Failure while parsing part[1]"));
+			assertEquals("Part[1] exceeded the disk usage limit of 5 bytes", ex.getCause().getMessage());
+		});
+	}
+
+	@Test
+	public void readPartHeadersTooBig() {
+		testMultipartExceptions(reader -> reader.setMaxInMemorySize(1), ex -> {
+			assertEquals(DecodingException.class, ex.getClass());
+			assertThat(ex.getMessage(), startsWith("Failure while parsing part[1]"));
+			assertEquals("Part[1] exceeded the in-memory limit of 1 bytes", ex.getCause().getMessage());
+		});
+	}
+
+	private void testMultipartExceptions(
+			Consumer<SynchronossPartHttpMessageReader> configurer, Consumer<Throwable> assertions) {
+
+		SynchronossPartHttpMessageReader reader = new SynchronossPartHttpMessageReader();
+		configurer.accept(reader);
+		MultipartHttpMessageReader multipartReader = new MultipartHttpMessageReader(reader);
+		StepVerifier.create(multipartReader.readMono(PARTS_ELEMENT_TYPE, generateMultipartRequest(), emptyMap()))
+				.consumeErrorWith(assertions)
+				.verify();
+	}
 
 	private ServerHttpRequest generateMultipartRequest() {
-
 		MultipartBodyBuilder partsBuilder = new MultipartBodyBuilder();
-		partsBuilder.part("fooPart", new ClassPathResource("org/springframework/http/codec/multipart/foo.txt"));
-		partsBuilder.part("barPart", "bar");
+		partsBuilder.part("filePart", new ClassPathResource("org/springframework/http/codec/multipart/foo.txt"));
+		partsBuilder.part("textPart", "sample-text");
 
 		MockClientHttpRequest outputMessage = new MockClientHttpRequest(HttpMethod.POST, "/");
 		new MultipartHttpMessageWriter()
 				.write(Mono.just(partsBuilder.build()), null, MediaType.MULTIPART_FORM_DATA, outputMessage, null)
 				.block(Duration.ofSeconds(5));
-
 		return MockServerHttpRequest.post("/")
 				.contentType(outputMessage.getHeaders().getContentType())
 				.body(outputMessage.getBody());
@@ -152,4 +207,12 @@ public class SynchronossPartHttpMessageReaderTests {
 				.body(Flux.just(new DefaultDataBufferFactory().wrap("invalid content".getBytes())));
 	}
 
+	private static class ZeroDemandSubscriber extends BaseSubscriber<MultiValueMap<String, Part>> {
+
+		@Override
+		protected void hookOnSubscribe(Subscription subscription) {
+			// Just subscribe without requesting
+		}
+	}
+
 }

spring-web/src/test/java/org/springframework/http/codec/support/ServerCodecConfigurerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -124,13 +124,41 @@ public class ServerCodecConfigurerTests {
 				.filter(e -> e == encoder).orElse(null));
 	}
 
+	@Test
+	public void maxInMemorySize() {
+		int size = 99;
+		this.configurer.defaultCodecs().maxInMemorySize(size);
+		List<HttpMessageReader<?>> readers = this.configurer.getReaders();
+		assertEquals(13, readers.size());
+		assertEquals(size, ((ByteArrayDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((ByteBufferDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((DataBufferDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((ResourceDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((StringDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((ProtobufDecoder) getNextDecoder(readers)).getMaxMessageSize());
+		assertEquals(size, ((FormHttpMessageReader) nextReader(readers)).getMaxInMemorySize());
+		assertEquals(size, ((SynchronossPartHttpMessageReader) nextReader(readers)).getMaxInMemorySize());
+
+		MultipartHttpMessageReader multipartReader = (MultipartHttpMessageReader) nextReader(readers);
+		SynchronossPartHttpMessageReader reader = (SynchronossPartHttpMessageReader) multipartReader.getPartReader();
+		assertEquals(size, (reader).getMaxInMemorySize());
+
+		assertEquals(size, ((Jackson2JsonDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((Jackson2SmileDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((Jaxb2XmlDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((StringDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+	}
 
 	private Decoder<?> getNextDecoder(List<HttpMessageReader<?>> readers) {
-		HttpMessageReader<?> reader = readers.get(this.index.getAndIncrement());
+		HttpMessageReader<?> reader = nextReader(readers);
 		assertEquals(DecoderHttpMessageReader.class, reader.getClass());
 		return ((DecoderHttpMessageReader<?>) reader).getDecoder();
 	}
 
+	private HttpMessageReader<?> nextReader(List<HttpMessageReader<?>> readers) {
+		return readers.get(this.index.getAndIncrement());
+	}
+
 	private Encoder<?> getNextEncoder(List<HttpMessageWriter<?>> writers) {
 		HttpMessageWriter<?> writer = writers.get(this.index.getAndIncrement());
 		assertEquals(EncoderHttpMessageWriter.class, writer.getClass());

spring-web/src/test/java/org/springframework/http/codec/xml/XmlEventDecoderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -28,8 +28,10 @@ import reactor.test.StepVerifier;
 
 import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author Arjen Poutsma
@@ -44,11 +46,12 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 
 	private XmlEventDecoder decoder = new XmlEventDecoder();
 
+
 	@Test
 	public void toXMLEventsAalto() {
 
 		Flux<XMLEvent> events =
-				this.decoder.decode(stringBuffer(XML), null, null, Collections.emptyMap());
+				this.decoder.decode(stringBufferMono(XML), null, null, Collections.emptyMap());
 
 		StepVerifier.create(events)
 				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
@@ -69,7 +72,7 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 		decoder.useAalto = false;
 
 		Flux<XMLEvent> events =
-				this.decoder.decode(stringBuffer(XML), null, null, Collections.emptyMap());
+				this.decoder.decode(stringBufferMono(XML), null, null, Collections.emptyMap());
 
 		StepVerifier.create(events)
 				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
@@ -86,10 +89,32 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 				.verify();
 	}
 
+	@Test
+	public void toXMLEventsWithLimit() {
+
+		this.decoder.setMaxInMemorySize(6);
+
+		Flux<String> source = Flux.just(
+				"<pojo>", "<foo>", "foofoo", "</foo>", "<bar>", "barbarbar", "</bar>", "</pojo>");
+
+		Flux<XMLEvent> events = this.decoder.decode(
+				source.map(this::stringBuffer), null, null, Collections.emptyMap());
+
+		StepVerifier.create(events)
+				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
+				.consumeNextWith(e -> assertStartElement(e, "pojo"))
+				.consumeNextWith(e -> assertStartElement(e, "foo"))
+				.consumeNextWith(e -> assertCharacters(e, "foofoo"))
+				.consumeNextWith(e -> assertEndElement(e, "foo"))
+				.consumeNextWith(e -> assertStartElement(e, "bar"))
+				.expectError(DataBufferLimitException.class)
+				.verify();
+	}
+
 	@Test
 	public void decodeErrorAalto() {
 		Flux<DataBuffer> source = Flux.concat(
-				stringBuffer("<pojo>"),
+				stringBufferMono("<pojo>"),
 				Flux.error(new RuntimeException()));
 
 		Flux<XMLEvent> events =
@@ -107,7 +132,7 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 		decoder.useAalto = false;
 
 		Flux<DataBuffer> source = Flux.concat(
-				stringBuffer("<pojo>"),
+				stringBufferMono("<pojo>"),
 				Flux.error(new RuntimeException()));
 
 		Flux<XMLEvent> events =
@@ -133,13 +158,15 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 		assertEquals(expectedData, event.asCharacters().getData());
 	}
 
-	private Mono<DataBuffer> stringBuffer(String value) {
-		return Mono.defer(() -> {
-			byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
-			DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
-			buffer.write(bytes);
-			return Mono.just(buffer);
-		});
+	private DataBuffer stringBuffer(String value) {
+		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+		buffer.write(bytes);
+		return buffer;
+	}
+
+	private Mono<DataBuffer> stringBufferMono(String value) {
+		return Mono.defer(() -> Mono.just(stringBuffer(value)));
 	}
 
 }