SSE: bound the line length and the event size

httpclient5-sse/src/main/java/org/apache/hc/client5/http/sse/EventSourceConfig.java

@@ -29,6 +29,7 @@
 import java.util.Objects;
 
 import org.apache.hc.client5.http.sse.impl.ExponentialJitterBackoff;
+import org.apache.hc.core5.util.Args;
 
 /**
  * Immutable configuration for {@link EventSource} behavior, primarily covering
@@ -43,6 +44,8 @@
  *       The strategy can also honor server hints (SSE {@code retry:} field or HTTP {@code Retry-After}).</li>
  *   <li>{@link #maxReconnects}: maximum number of reconnect attempts before giving up.
  *       A value of {@code -1} means unlimited attempts.</li>
+ *   <li>{@link #maxLineLength}: maximum length of a single line of the stream.</li>
+ *   <li>{@link #maxEventSize}: maximum size of the data accumulated for a single event.</li>
  * </ul>
  *
  * <h3>Thread-safety</h3>
@@ -74,20 +77,55 @@ public final class EventSourceConfig {
      */
     public final int maxReconnects;
 
+    /**
+     * Default value of {@link #maxLineLength}, 1 MiB.
+     *
+     * @since 5.7
+     */
+    public static final int DEFAULT_MAX_LINE_LENGTH = 1024 * 1024;
+
+    /**
+     * Default value of {@link #maxEventSize}, 8 MiB.
+     *
+     * @since 5.7
+     */
+    public static final int DEFAULT_MAX_EVENT_SIZE = 8 * 1024 * 1024;
+
+    /**
+     * Maximum length of a single line, excluding the line terminator. Counted in characters
+     * by the char parser and in bytes by the byte parser. A longer line fails the stream.
+     *
+     * @since 5.7
+     */
+    public final int maxLineLength;
+
+    /**
+     * Maximum size of the {@code data:} content accumulated for one event before the
+     * terminating blank line. Counted in characters. A larger event fails the stream.
+     *
+     * @since 5.7
+     */
+    public final int maxEventSize;
+
     /**
      * Default configuration:
      * <ul>
      *   <li>{@link #backoff} = {@code new ExponentialJitterBackoff(1000, 30000, 2.0, 250)}</li>
      *   <li>{@link #maxReconnects} = {@code -1} (unlimited)</li>
+     *   <li>{@link #maxLineLength} = 1 MiB</li>
+     *   <li>{@link #maxEventSize} = 8 MiB</li>
      * </ul>
      *
      * @since 5.7
      */
     public static final EventSourceConfig DEFAULT = builder().build();
 
-    private EventSourceConfig(final BackoffStrategy backoff, final int maxReconnects) {
+    private EventSourceConfig(final BackoffStrategy backoff, final int maxReconnects,
+                              final int maxLineLength, final int maxEventSize) {
         this.backoff = backoff;
         this.maxReconnects = maxReconnects;
+        this.maxLineLength = maxLineLength;
+        this.maxEventSize = maxEventSize;
     }
 
     /**
@@ -111,6 +149,8 @@ public static final class Builder {
         private BackoffStrategy backoff =
                 new ExponentialJitterBackoff(1_000L, 30_000L, 2.0, 250L);
         private int maxReconnects = -1;
+        private int maxLineLength = DEFAULT_MAX_LINE_LENGTH;
+        private int maxEventSize = DEFAULT_MAX_EVENT_SIZE;
 
         /**
          * Sets the reconnect/backoff strategy.
@@ -139,14 +179,38 @@ public Builder maxReconnects(final int nmaxReconnects) {
             return this;
         }
 
+        /**
+         * Sets the maximum length of a single line.
+         *
+         * @param maxLineLength maximum line length, must be positive
+         * @return this builder
+         * @since 5.7
+         */
+        public Builder maxLineLength(final int maxLineLength) {
+            this.maxLineLength = Args.positive(maxLineLength, "Max line length");
+            return this;
+        }
+
+        /**
+         * Sets the maximum size of the data accumulated for a single event.
+         *
+         * @param maxEventSize maximum event size, must be positive
+         * @return this builder
+         * @since 5.7
+         */
+        public Builder maxEventSize(final int maxEventSize) {
+            this.maxEventSize = Args.positive(maxEventSize, "Max event size");
+            return this;
+        }
+
         /**
          * Builds an immutable {@link EventSourceConfig}.
          *
          * @return the config
          * @since 5.7
          */
         public EventSourceConfig build() {
-            return new EventSourceConfig(backoff, maxReconnects);
+            return new EventSourceConfig(backoff, maxReconnects, maxLineLength, maxEventSize);
         }
     }
 }

httpclient5-sse/src/main/java/org/apache/hc/client5/http/sse/impl/ByteSseEntityConsumer.java

@@ -30,10 +30,13 @@
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 
+import org.apache.hc.client5.http.sse.EventSourceConfig;
 import org.apache.hc.core5.annotation.Internal;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.MessageConstraintException;
 import org.apache.hc.core5.http.nio.entity.AbstractBinAsyncEntityConsumer;
+import org.apache.hc.core5.util.Args;
 
 /**
  * Low-allocation binary consumer for Server-Sent Events (SSE).
@@ -52,6 +55,8 @@
  *   <li>Implements WHATWG SSE fields: {@code data}, {@code id}, {@code event}, {@code retry}.
  *       Unknown fields and malformed {@code retry} values are ignored.</li>
  *  <li>At end of stream, discards any incomplete line or event.</li>
+ *  <li>Fails the stream with {@link MessageConstraintException} if a line exceeds the maximum
+ *      line length in bytes or an event accumulates more than the maximum event size.</li>
  * </ul>
  *
  * <h3>Thread-safety</h3>
@@ -71,6 +76,8 @@ public final class ByteSseEntityConsumer extends AbstractBinAsyncEntityConsumer<
     private static final byte SPACE = (byte) ' ';
 
     private final SseCallbacks cb;
+    private final int maxLineLength;
+    private final int maxEventSize;
 
     // line accumulator
     private byte[] lineBuf = new byte[256];
@@ -88,7 +95,13 @@ public final class ByteSseEntityConsumer extends AbstractBinAsyncEntityConsumer<
     private boolean bomDone = false; // once true, no further BOM detection
 
     public ByteSseEntityConsumer(final SseCallbacks callbacks) {
+        this(callbacks, EventSourceConfig.DEFAULT_MAX_LINE_LENGTH, EventSourceConfig.DEFAULT_MAX_EVENT_SIZE);
+    }
+
+    public ByteSseEntityConsumer(final SseCallbacks callbacks, final int maxLineLength, final int maxEventSize) {
         this.cb = callbacks;
+        this.maxLineLength = Args.positive(maxLineLength, "Max line length");
+        this.maxEventSize = Args.positive(maxEventSize, "Max event size");
     }
 
     @Override
@@ -101,7 +114,16 @@ public void streamStart(final ContentType contentType) throws HttpException, IOE
     }
 
     @Override
-    protected void data(final ByteBuffer src, final boolean endOfStream) {
+    protected void data(final ByteBuffer src, final boolean endOfStream) throws IOException {
+        try {
+            consumeBytes(src, endOfStream);
+        } catch (final MessageConstraintException ex) {
+            releaseResources();
+            throw ex;
+        }
+    }
+
+    private void consumeBytes(final ByteBuffer src, final boolean endOfStream) throws IOException {
         if (!bomDone) {
             while (src.hasRemaining() && bomMatched < 3) {
                 final int expected = (bomMatched == 0) ? 0xEF : (bomMatched == 1 ? 0xBB : 0xBF);
@@ -142,7 +164,7 @@ protected void data(final ByteBuffer src, final boolean endOfStream) {
         }
     }
 
-    private void processByte(final byte b) {
+    private void processByte(final byte b) throws IOException {
         if (b == LF) {
             if (pendingCr) {
                 // LF completing a CRLF pair; the line was already emitted on the CR.
@@ -168,7 +190,10 @@ private void flushEndOfStream() {
         type = null;
     }
 
-    private void appendByte(final byte b) {
+    private void appendByte(final byte b) throws MessageConstraintException {
+        if (lineLen >= maxLineLength) {
+            throw new MessageConstraintException("Maximum line length exceeded: " + maxLineLength);
+        }
         ensureCapacity(lineLen + 1);
         lineBuf[lineLen++] = b;
     }
@@ -189,13 +214,14 @@ public void releaseResources() {
         lineLen = 0;
         pendingCr = false;
         data.setLength(0);
+        data.trimToSize();
         id = null;
         type = null;
         bomMatched = 0;
         bomDone = false;
     }
 
-    private void handleLine(final byte[] buf, final int len) {
+    private void handleLine(final byte[] buf, final int len) throws IOException {
         if (len == 0) {
             dispatch();
             return;
@@ -222,6 +248,9 @@ private void handleLine(final byte[] buf, final int len) {
         // Compare ASCII field name without allocations
         if (fLen == 4 && buf[0] == 'd' && buf[1] == 'a' && buf[2] == 't' && buf[3] == 'a') {
             final String v = new String(buf, vStart, len - vStart, StandardCharsets.UTF_8);
+            if ((long) data.length() + v.length() + 1 > maxEventSize) {
+                throw new MessageConstraintException("Maximum event size exceeded: " + maxEventSize);
+            }
             data.append(v).append('\n');
         } else if (fLen == 5 && buf[0] == 'e' && buf[1] == 'v' && buf[2] == 'e' && buf[3] == 'n' && buf[4] == 't') {
             type = new String(buf, vStart, len - vStart, StandardCharsets.UTF_8);

httpclient5-sse/src/main/java/org/apache/hc/client5/http/sse/impl/DefaultEventSource.java

@@ -119,6 +119,8 @@ public final class DefaultEventSource implements EventSource {
     private final Executor callbackExecutor;
     private final BackoffStrategy backoff;
     private final int maxReconnects;
+    private final int maxLineLength;
+    private final int maxEventSize;
     private final SseParser parser;
 
     private final AtomicBoolean started = new AtomicBoolean(false);
@@ -195,6 +197,8 @@ public DefaultEventSource(final CloseableHttpAsyncClient client,
         final EventSourceConfig cfg = (config != null) ? config : EventSourceConfig.DEFAULT;
         this.backoff = cfg.backoff;
         this.maxReconnects = cfg.maxReconnects;
+        this.maxLineLength = cfg.maxLineLength;
+        this.maxEventSize = cfg.maxEventSize;
         this.parser = parser != null ? parser : SseParser.CHAR;
     }
 
@@ -414,8 +418,8 @@ public void onRetry(final long retryMs) {
         };
 
         final AsyncEntityConsumer<Void> entity =
-                (parser == SseParser.BYTE) ? new ByteSseEntityConsumer(cbs)
-                        : new SseEntityConsumer(cbs);
+                (parser == SseParser.BYTE) ? new ByteSseEntityConsumer(cbs, maxLineLength, maxEventSize)
+                        : new SseEntityConsumer(cbs, maxLineLength, maxEventSize);
 
         return new SseResponseConsumer(entity, ms -> retryAfterHintMs = Math.max(0L, ms));
     }

httpclient5-sse/src/main/java/org/apache/hc/client5/http/sse/impl/ServerSentEventReader.java

@@ -26,7 +26,12 @@
  */
 package org.apache.hc.client5.http.sse.impl;
 
+import java.io.IOException;
+
+import org.apache.hc.client5.http.sse.EventSourceConfig;
 import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.http.MessageConstraintException;
+import org.apache.hc.core5.util.Args;
 
 /**
  * Minimal-allocation SSE line parser.
@@ -54,15 +59,27 @@ default void onLastEventId(final String id) {
     }
 
     private final Callback cb;
+    private final int maxEventSize;
     private final StringBuilder data = new StringBuilder(128);
     private String type; // defaults to "message"
     private String id;
 
     public ServerSentEventReader(final Callback cb) {
+        this(cb, EventSourceConfig.DEFAULT_MAX_EVENT_SIZE);
+    }
+
+    public ServerSentEventReader(final Callback cb, final int maxEventSize) {
         this.cb = cb;
+        this.maxEventSize = Args.positive(maxEventSize, "Max event size");
     }
 
-    public void line(final String line) {
+    /**
+     * Processes one line without its terminator.
+     *
+     * @throws MessageConstraintException if the data accumulated for the current event
+     *                                    exceeds the maximum event size
+     */
+    public void line(final String line) throws IOException {
         // Trim possible trailing CR without reallocating
         final int L0 = line.length();
         int end = L0;
@@ -114,6 +131,11 @@ public void line(final String line) {
 
             // data: <value> (append newline; removed on dispatch)
             if (vStart <= end) {
+                if ((long) data.length() + (end - vStart) + 1 > maxEventSize) {
+                    data.setLength(0);
+                    data.trimToSize();
+                    throw new MessageConstraintException("Maximum event size exceeded: " + maxEventSize);
+                }
                 data.append(line, vStart, end).append('\n');
             }