diff --git a/changes.md b/changes.md index b80f352af9..66450c3303 100644 --- a/changes.md +++ b/changes.md @@ -131,6 +131,9 @@ Changes log `MediaType.MULTIPART_MIXED` constant. Issue #1587. - `EventStreamRepresentation` sends the response head before the first event, so that clients get the status and headers of an idle stream without heartbeats. Issue #1598. + - `EventStreamReader` limits the size of an event, 1 Mi characters by default + (`setMaxEventSize`). A larger line or event fails the read and closes the reader. Issue + #1597. - Added the experimental `org.restlet.ext.websocket` extension, on HTTP/1.1 upgrades. `WebSocketUpgrader` (or `WebSocketUpgrade.accept`) accepts upgrades that went through the normal routing, filters and authenticators, and runs a `WebSocketHandler` on its own diff --git a/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java b/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java index 177d9de104..4edd359553 100644 --- a/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java +++ b/org.restlet/src/main/java/org/restlet/representation/EventStreamReader.java @@ -24,9 +24,15 @@ * A blocking reader of Server-Sent Events. When created with an {@link EventStreamSource}, it * reconnects after EOF, honoring the server's {@code retry:} field and passing the latest event ID * to the source for use in a {@code Last-Event-ID} request header. + * + *

The size of an event is limited (see {@link #setMaxEventSize(int)}), so that a server can't + * make the reader buffer an unbounded line or event in memory. */ public final class EventStreamReader implements Iterator, AutoCloseable { + /** The default maximum size of an event, in characters. */ + public static final int DEFAULT_MAX_EVENT_SIZE = 1024 * 1024; + private static final long DEFAULT_RETRY_MS = 3000; private final EventStreamSource source; @@ -41,8 +47,12 @@ public final class EventStreamReader implements Iterator, AutoC private volatile long retryDelayMs = DEFAULT_RETRY_MS; + private volatile int maxEventSize = DEFAULT_MAX_EVENT_SIZE; + private boolean beginning = true; + private boolean skipLineFeed; + private boolean currentEof; private boolean sourceStopped; @@ -104,6 +114,31 @@ public long getRetryDelayMs() { return retryDelayMs; } + /** + * Returns the maximum size of an event: the number of characters of its lines, without line + * terminators. Defaults to {@link #DEFAULT_MAX_EVENT_SIZE}. + * + * @return The maximum size of an event, in characters. + */ + public int getMaxEventSize() { + return maxEventSize; + } + + /** + * Sets the maximum size of an event: the number of characters of its lines, without line + * terminators. When an event exceeds it, {@link #readEvent()} throws an {@link IOException} and + * the reader is closed. + * + * @param maxEventSize The maximum size of an event, in characters. + * @throws IllegalArgumentException If the size is not positive. + */ + public void setMaxEventSize(int maxEventSize) { + if (maxEventSize <= 0) { + throw new IllegalArgumentException("The maximum event size must be positive"); + } + this.maxEventSize = maxEventSize; + } + /** * Reads the next event, blocking as needed. * @@ -160,25 +195,66 @@ public void close() throws IOException { private ServerSentEvent readCurrentEvent() throws IOException { PendingEvent pending = new PendingEvent(); ServerSentEvent result = null; + int size = 0; while (result == null && !closed) { - String line = reader.readLine(); + String line = readLine(maxEventSize - size); if (line == null) { currentEof = true; return pending.build(); } - result = processLine(stripBom(line), pending); + size = line.isEmpty() ? 0 : size + line.length(); + result = processLine(line, pending); } return result; } - private String stripBom(String line) { - if (beginning) { - beginning = false; - if (!line.isEmpty() && line.charAt(0) == '\uFEFF') { - return line.substring(1); + /** + * Reads a line ended by CRLF, LF or CR, like {@link BufferedReader#readLine()}, but no longer + * than the given number of characters. A line feed following a carriage return is skipped on + * the next read, so that a CR doesn't block until the next character arrives. The optional byte + * order mark at the start of the stream is skipped, and not counted in the limit. + */ + private String readLine(int limit) throws IOException { + StringBuilder line = new StringBuilder(); + while (true) { + int c = reader.read(); + if (skipLineFeed) { + skipLineFeed = false; + if (c == '\n') { + continue; + } + } + if (beginning) { + beginning = false; + if (c == '\uFEFF') { + continue; + } + } + if (c < 0) { + return line.isEmpty() ? null : line.toString(); + } + if (c == '\n') { + return line.toString(); + } + if (c == '\r') { + skipLineFeed = true; + return line.toString(); + } + if (line.length() >= limit) { + IOException failure = + new IOException( + "The event exceeds the maximum size of " + + maxEventSize + + " characters"); + try { + close(); + } catch (IOException e) { + failure.addSuppressed(e); + } + throw failure; } + line.append((char) c); } - return line; } private ServerSentEvent processLine(String line, PendingEvent pending) { @@ -251,6 +327,7 @@ private boolean reconnect() throws IOException { } reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); beginning = true; + skipLineFeed = false; currentEof = false; return true; } diff --git a/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java b/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java index 689ea81283..f6914ca399 100644 --- a/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java +++ b/org.restlet/src/test/java/org/restlet/representation/EventStreamTestCase.java @@ -11,13 +11,16 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.FilterReader; import java.io.IOException; import java.io.InputStream; +import java.io.Reader; import java.io.StringReader; import java.nio.charset.StandardCharsets; import java.time.Duration; @@ -148,6 +151,96 @@ void reconnectsAfterRetryAndPassesTheLastEventId() throws IOException { reader.close(); } + @Test + void readsCarriageReturnLineEndingsWithoutWaitingForMoreInput() throws IOException { + // Fails instead of blocking when asked for more than the event, as a live stream would + Reader input = + new StringReader("data:a\rdata:b\r\r") { + @Override + public int read(char[] buffer, int offset, int length) throws IOException { + int count = super.read(buffer, offset, length); + if (count < 0) { + throw new AssertionError("Read past the available input"); + } + return count; + } + }; + EventStreamReader reader = new EventStreamReader(input); + + assertEquals("a\nb", reader.readEvent().getData()); + reader.close(); + } + + @Test + void acceptsEventsUpToTheMaximumSize() throws IOException { + EventStreamReader reader = + new EventStreamReader(new StringReader("data:1\nid:5\n\ndata:12345\n\n")); + reader.setMaxEventSize(10); + + assertEquals(10, reader.getMaxEventSize()); + assertEquals("1", reader.readEvent().getData()); + assertEquals("12345", reader.readEvent().getData()); + assertNull(reader.readEvent()); + reader.close(); + } + + @Test + void rejectsAnOversizedLineAndStopsReading() throws IOException { + EventStreamReader reader = + new EventStreamReader(new StringReader("data:12345678901\n\ndata:next\n\n")); + reader.setMaxEventSize(10); + + IOException failure = assertThrows(IOException.class, reader::readEvent); + + assertTrue(failure.getMessage().contains("10 characters")); + assertNull(reader.readEvent()); + assertFalse(reader.hasNext()); + } + + @Test + void rejectsAnOversizedEventSpreadOverLines() throws IOException { + EventStreamReader reader = new EventStreamReader(new StringReader(":12345\n:12345\n\n")); + reader.setMaxEventSize(10); + + assertThrows(IOException.class, reader::readEvent); + } + + @Test + void doesNotCountTheByteOrderMarkInTheEventSize() throws IOException { + EventStreamReader reader = new EventStreamReader(new StringReader("\uFEFFdata:12345\n\n")); + reader.setMaxEventSize(10); + + assertEquals("12345", reader.readEvent().getData()); + reader.close(); + } + + @Test + void keepsTheSizeFailureWhenClosingFails() { + IOException closeFailure = new IOException("Close failed"); + EventStreamReader reader = + new EventStreamReader( + new FilterReader(new StringReader("data:12345678901\n\n")) { + @Override + public void close() throws IOException { + throw closeFailure; + } + }); + reader.setMaxEventSize(10); + + IOException failure = assertThrows(IOException.class, reader::readEvent); + + assertTrue(failure.getMessage().contains("10 characters")); + assertSame(closeFailure, failure.getSuppressed()[0]); + } + + @Test + void validatesTheMaximumEventSize() throws IOException { + try (EventStreamReader reader = new EventStreamReader(new StringReader(""))) { + assertEquals(EventStreamReader.DEFAULT_MAX_EVENT_SIZE, reader.getMaxEventSize()); + assertThrows(IllegalArgumentException.class, () -> reader.setMaxEventSize(0)); + } + } + @Test void validatesEventFields() { ServerSentEvent.Builder builder = ServerSentEvent.builder();