Add configurable SSE parser buffer limit

src/libraries/System.Net.ServerSentEvents/ref/System.Net.ServerSentEvents.cs

@@ -28,6 +28,13 @@ public static partial class SseParser
         public const string EventTypeDefault = "message";
         public static System.Net.ServerSentEvents.SseParser<string> Create(System.IO.Stream sseStream) { throw null; }
         public static System.Net.ServerSentEvents.SseParser<T> Create<T>(System.IO.Stream sseStream, System.Net.ServerSentEvents.SseItemParser<T> itemParser) { throw null; }
+        public static System.Net.ServerSentEvents.SseParser<T> Create<T>(System.IO.Stream sseStream, System.Net.ServerSentEvents.SseParserOptions<T> options) { throw null; }
+    }
+    public sealed partial class SseParserOptions<T>
+    {
+        public SseParserOptions(System.Net.ServerSentEvents.SseItemParser<T> itemParser) { }
+        public System.Net.ServerSentEvents.SseItemParser<T> ItemParser { get { throw null; } }
+        public int MaxBufferSize { get { throw null; } set { } }
     }
     public sealed partial class SseParser<T>
     {

src/libraries/System.Net.ServerSentEvents/src/System.Net.ServerSentEvents.csproj

@@ -20,6 +20,7 @@ System.Net.ServerSentEvents.SseParser</PackageDescription>
     <Compile Include="System\Net\ServerSentEvents\SseItem.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseItemParser.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseParser.cs" />
+    <Compile Include="System\Net\ServerSentEvents\SseParserOptions.cs" />
     <Compile Include="System\Net\ServerSentEvents\ThrowHelper.cs" />
   </ItemGroup>
 

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser.cs

@@ -3,7 +3,6 @@
 
 using System.IO;
 using System.Text;
-
 namespace System.Net.ServerSentEvents
 {
     /// <summary>Provides a parser for parsing server-sent events.</summary>
@@ -32,19 +31,21 @@ public static SseParser<string> Create(Stream sseStream) =>
         /// <param name="itemParser">The parser to use to transform each payload of bytes into a data element.</param>
         /// <returns>The enumerable, which can be enumerated synchronously or asynchronously.</returns>
         /// <exception cref="ArgumentNullException"><paramref name="sseStream"/> or <paramref name="itemParser"/> is null.</exception>
-        public static SseParser<T> Create<T>(Stream sseStream, SseItemParser<T> itemParser)
-        {
-            if (sseStream is null)
-            {
-                ThrowHelper.ThrowArgumentNullException(nameof(sseStream));
-            }
+        public static SseParser<T> Create<T>(Stream sseStream, SseItemParser<T> itemParser) =>
+            Create(sseStream, new SseParserOptions<T>(itemParser));
 
-            if (itemParser is null)
-            {
-                ThrowHelper.ThrowArgumentNullException(nameof(itemParser));
-            }
+        /// <summary>Creates a parser for parsing a <paramref name="sseStream"/> of server-sent events into a sequence of <see cref="SseItem{T}"/> values.</summary>
+        /// <typeparam name="T">Specifies the type of data in each event.</typeparam>
+        /// <param name="sseStream">The stream containing the data to parse.</param>
+        /// <param name="options">The options to use when parsing the stream.</param>
+        /// <returns>The enumerable, which can be enumerated synchronously or asynchronously.</returns>
+        /// <exception cref="ArgumentNullException"><paramref name="sseStream"/> or <paramref name="options"/> is null.</exception>
+        public static SseParser<T> Create<T>(Stream sseStream, SseParserOptions<T> options)
+        {
+            ArgumentNullException.ThrowIfNull(sseStream);
+            ArgumentNullException.ThrowIfNull(options);
 
-            return new SseParser<T>(sseStream, itemParser);
+            return new SseParser<T>(sseStream, options);
         }
     }
 }

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParserOptions.cs

@@ -0,0 +1,37 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+namespace System.Net.ServerSentEvents
+{
+    /// <summary>Provides options for parsing server-sent events.</summary>
+    /// <typeparam name="T">Specifies the type of data parsed from an event.</typeparam>
+    public sealed class SseParserOptions<T>
+    {
+        /// <summary>Initializes a new instance of the <see cref="SseParserOptions{T}"/> class.</summary>
+        /// <param name="itemParser">The parser to use to transform each payload of bytes into a data element.</param>
+        /// <exception cref="ArgumentNullException"><paramref name="itemParser"/> is null.</exception>
+        public SseParserOptions(SseItemParser<T> itemParser)
+        {
+            ArgumentNullException.ThrowIfNull(itemParser);
+
+            ItemParser = itemParser;
+        }
+
+        /// <summary>Gets the parser to use to transform each payload of bytes into a data element.</summary>
+        public SseItemParser<T> ItemParser { get; }
+
+        /// <summary>Gets or sets the maximum buffer size, or -1 to use the default limit.</summary>
+        /// <exception cref="ArgumentOutOfRangeException">The value set is less than -1.</exception>
+        public int MaxBufferSize
+        {
+            get => _maxBufferSize;
+            set
+            {
+                ArgumentOutOfRangeException.ThrowIfLessThan(value, -1);
+                _maxBufferSize = value;
+            }
+        }
+
+        private int _maxBufferSize = -1;
+    }
+}

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseParser_1.cs

@@ -41,6 +41,9 @@ public sealed class SseParser<T>
             1024;
 #endif
 
+        /// <summary>The maximum amount of data buffered by default.</summary>
+        private const int DefaultMaxBufferSize = 1024 * 1024 * 1024;
+
         /// <summary>The stream to be parsed.</summary>
         private readonly Stream _stream;
         /// <summary>The parser delegate used to transform bytes into a <typeparamref name="T"/>.</summary>
@@ -74,7 +77,7 @@ public sealed class SseParser<T>
         /// <remarks>This can be different than <see cref="_dataLength"/> != 0 if empty data was appended.</remarks>
         private bool _dataAppended;
 
-        private int _maxBufferSize = 1024 * 1024 * 1024;
+        private readonly int _maxBufferSize;
 
         /// <summary>The event type for the next event.</summary>
         private string? _eventType;
@@ -87,11 +90,12 @@ public sealed class SseParser<T>
 
         /// <summary>Initialize the enumerable.</summary>
         /// <param name="stream">The stream to parse.</param>
-        /// <param name="itemParser">The function to use to parse payload bytes into a <typeparamref name="T"/>.</param>
-        internal SseParser(Stream stream, SseItemParser<T> itemParser)
+        /// <param name="options">The options to use to parse the stream.</param>
+        internal SseParser(Stream stream, SseParserOptions<T> options)
         {
             _stream = stream;
-            _itemParser = itemParser;
+            _itemParser = options.ItemParser;
+            _maxBufferSize = options.MaxBufferSize == -1 ? DefaultMaxBufferSize : options.MaxBufferSize;
         }
 
         /// <summary>Gets an enumerable of the server-sent events from this parser.</summary>
@@ -104,7 +108,7 @@ public IEnumerable<SseItem<T>> Enumerate()
             // Rent a line buffer. This will grow as needed. The line buffer is what's passed to the stream,
             // so we want it to be large enough to reduce the number of reads we need to do when data is
             // arriving quickly. (In debug, we use a smaller buffer to stress the growth and shifting logic.)
-            _lineBuffer = ArrayPool<byte>.Shared.Rent(DefaultArrayPoolRentSize);
+            _lineBuffer = ArrayPool<byte>.Shared.Rent(Math.Min(DefaultArrayPoolRentSize, _maxBufferSize));
             try
             {
                 // Spec: "Event streams in this format must always be encoded as UTF-8".
@@ -184,7 +188,7 @@ public async IAsyncEnumerable<SseItem<T>> EnumerateAsync([EnumeratorCancellation
             // Rent a line buffer. This will grow as needed. The line buffer is what's passed to the stream,
             // so we want it to be large enough to reduce the number of reads we need to do when data is
             // arriving quickly. (In debug, we use a smaller buffer to stress the growth and shifting logic.)
-            _lineBuffer = ArrayPool<byte>.Shared.Rent(DefaultArrayPoolRentSize);
+            _lineBuffer = ArrayPool<byte>.Shared.Rent(Math.Min(DefaultArrayPoolRentSize, _maxBufferSize));
             try
             {
                 // Spec: "Event streams in this format must always be encoded as UTF-8".
@@ -306,14 +310,24 @@ private void ShiftOrGrowLineBufferIfNecessary()
                 }
                 else if (_lineLength == _lineBuffer.Length)
                 {
+                    if (_lineLength >= _maxBufferSize)
+                    {
+                        throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength);
+                    }
+
                     int newLength;
                     try
                     {
                         newLength = checked(_lineBuffer.Length * 2);
                     }
                     catch (OverflowException)
                     {
-                        throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength);
+                        newLength = int.MaxValue;
+                    }
+
+                    if (newLength > _maxBufferSize)
+                    {
+                        newLength = _maxBufferSize;
                     }
 
                     GrowBuffer(ref _lineBuffer, newLength);
@@ -507,6 +521,10 @@ private int FillLineBuffer()
             if (bytesRead > 0)
             {
                 _lineLength += bytesRead;
+                if (_lineLength > _maxBufferSize)
+                {
+                    throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength);
+                }
             }
             else
             {
@@ -528,6 +546,10 @@ private async ValueTask<int> FillLineBufferAsync(CancellationToken cancellationT
             if (bytesRead > 0)
             {
                 _lineLength += bytesRead;
+                if (_lineLength > _maxBufferSize)
+                {
+                    throw new InvalidDataException(SR.InvalidDataException_SseExceededMaxLength);
+                }
             }
             else
             {

src/libraries/System.Net.ServerSentEvents/tests/SseParserTests.cs

@@ -6,7 +6,6 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
-using System.Reflection;
 using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization;
@@ -23,7 +22,18 @@ public void Parse_InvalidArguments_Throws()
         {
             AssertExtensions.Throws<ArgumentNullException>("sseStream", () => SseParser.Create(null));
             AssertExtensions.Throws<ArgumentNullException>("sseStream", () => SseParser.Create(null, delegate { return ""; }));
-            AssertExtensions.Throws<ArgumentNullException>("itemParser", () => SseParser.Create<string>(Stream.Null, null));
+            AssertExtensions.Throws<ArgumentNullException>("itemParser", () => SseParser.Create<string>(Stream.Null, (SseItemParser<string>)null));
+            AssertExtensions.Throws<ArgumentNullException>("itemParser", () => new SseParserOptions<string>(null));
+            AssertExtensions.Throws<ArgumentNullException>("sseStream", () => SseParser.Create<string>(null, new SseParserOptions<string>(delegate { return ""; })));
+            AssertExtensions.Throws<ArgumentNullException>("options", () => SseParser.Create<string>(Stream.Null, (SseParserOptions<string>)null));
+        }
+
+        [Fact]
+        public void Options_DefaultMaxBufferSize()
+        {
+            var options = new SseParserOptions<string>(delegate { return ""; });
+
+            Assert.Equal(-1, options.MaxBufferSize);
         }
 
         [Fact]
@@ -905,18 +915,12 @@ public async Task ArrayPoolRental_Closure(string newline, bool trickle, bool use
         [MemberData(nameof(NewlineAsyncData))]
         public async Task Parse_LongLineCap_Throws(string newline, bool useAsync)
         {
-            // Temporary workaround until we expose limit in public API
-            void ReduceLineLengthLimit(SseParser<string> parser)
-            {
-                Type type = typeof(SseParser<string>);
-                var field = type.GetField("_maxBufferSize", BindingFlags.Instance | BindingFlags.NonPublic);
-                Assert.NotNull(field);
-                field.SetValue(parser, 10 * 1024);
-            }
-
             using Stream stream = new InfiniteLineStream($"data: shortline{newline}{newline}data: ");
-            var parser = SseParser.Create(stream);
-            ReduceLineLengthLimit(parser);
+            var options = new SseParserOptions<string>(static (_, bytes) => Encoding.UTF8.GetString(bytes.ToArray()))
+            {
+                MaxBufferSize = 10 * 1024
+            };
+            var parser = SseParser.Create(stream, options);
 
             if (useAsync)
             {