From ef84a8383a8887f79efdcb4610d104d2a41d1f53 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Wed, 28 Oct 2026 15:20:10 +0800 Subject: [PATCH 1/4] Enforce receive message size limits during decompression Bound decompressed request and response messages by MaxReceiveMessageSize before deserialization. Share bounded copying between client and server and cover compressed unary and streaming messages with regression tests. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../GrpcServiceOptions.cs | 1 + .../Internal/PipeExtensions.cs | 12 +- src/Grpc.Net.Client/GrpcChannelOptions.cs | 1 + .../Internal/StreamExtensions.cs | 17 +- src/Shared/CommonGrpcProtocolHelpers.cs | 36 ++++ .../Server/MaxMessageSizeTests.cs | 29 +++ .../PipeExtensionsTestsBase.cs | 187 ++++++++++++++++++ .../MaximumMessageSizeTests.cs | 141 +++++++++++++ 8 files changed, 410 insertions(+), 14 deletions(-) diff --git a/src/Grpc.AspNetCore.Server/GrpcServiceOptions.cs b/src/Grpc.AspNetCore.Server/GrpcServiceOptions.cs index bb79664d5..cd495ce17 100644 --- a/src/Grpc.AspNetCore.Server/GrpcServiceOptions.cs +++ b/src/Grpc.AspNetCore.Server/GrpcServiceOptions.cs @@ -69,6 +69,7 @@ public bool MaxSendMessageSizeSpecified /// /// Gets or sets the maximum message size in bytes that can be received by the server. /// If the server receives a message that exceeds this limit, it throws an exception. + /// For compressed messages, the limit applies to both the compressed and decompressed message sizes. /// /// A null value removes the maximum message size limit. Defaults to 4,194,304 (4 MB). /// diff --git a/src/Grpc.AspNetCore.Server/Internal/PipeExtensions.cs b/src/Grpc.AspNetCore.Server/Internal/PipeExtensions.cs index e1f23326f..afe4192c2 100644 --- a/src/Grpc.AspNetCore.Server/Internal/PipeExtensions.cs +++ b/src/Grpc.AspNetCore.Server/Internal/PipeExtensions.cs @@ -23,6 +23,7 @@ using System.Runtime.CompilerServices; using Grpc.Core; using Grpc.Net.Compression; +using Grpc.Shared; using Microsoft.Extensions.Logging; namespace Grpc.AspNetCore.Server.Internal; @@ -406,7 +407,7 @@ private static bool TryReadMessage(ref ReadOnlySequence buffer, HttpContex } // Performance improvement would be to decompress without converting to an intermediary byte array - if (!TryDecompressMessage(context.Logger, encoding, context.Options.CompressionProviders, messageBuffer, out var decompressedMessage)) + if (!TryDecompressMessage(context.Logger, encoding, context.Options.CompressionProviders, messageBuffer, context.Options.MaxReceiveMessageSize, out var decompressedMessage)) { // https://github.com/grpc/grpc/blob/master/doc/compression.md#test-cases // A message compressed by a client in a way not supported by its server MUST fail with status UNIMPLEMENTED, @@ -439,19 +440,20 @@ private static bool TryReadMessage(ref ReadOnlySequence buffer, HttpContex return true; } - private static bool TryDecompressMessage(ILogger logger, string compressionEncoding, IReadOnlyDictionary compressionProviders, in ReadOnlySequence messageData, out ReadOnlySequence result) + private static bool TryDecompressMessage(ILogger logger, string compressionEncoding, IReadOnlyDictionary compressionProviders, in ReadOnlySequence messageData, int? maxReceiveMessageSize, out ReadOnlySequence result) { if (compressionProviders.TryGetValue(compressionEncoding, out var compressionProvider)) { GrpcServerLog.DecompressingMessage(logger, compressionProvider.EncodingName); - var output = new MemoryStream(); using (var compressionStream = compressionProvider.CreateDecompressionStream(new ReadOnlySequenceStream(messageData))) { - compressionStream.CopyTo(output); + if (!CommonGrpcProtocolHelpers.TryReadCompressedMessage(compressionStream, maxReceiveMessageSize, out result)) + { + throw new RpcException(ReceivedMessageExceedsLimitStatus); + } } - result = new ReadOnlySequence(output.GetBuffer(), 0, (int)output.Length); return true; } diff --git a/src/Grpc.Net.Client/GrpcChannelOptions.cs b/src/Grpc.Net.Client/GrpcChannelOptions.cs index b7adebe86..10b61e182 100644 --- a/src/Grpc.Net.Client/GrpcChannelOptions.cs +++ b/src/Grpc.Net.Client/GrpcChannelOptions.cs @@ -62,6 +62,7 @@ public sealed class GrpcChannelOptions /// /// Gets or sets the maximum message size in bytes that can be received by the client. If the client receives a /// message that exceeds this limit, it throws an exception. + /// For compressed messages, the limit applies to both the compressed and decompressed message sizes. /// /// A null value removes the maximum message size limit. Defaults to 4,194,304 (4 MB). /// diff --git a/src/Grpc.Net.Client/Internal/StreamExtensions.cs b/src/Grpc.Net.Client/Internal/StreamExtensions.cs index 0cf84e35c..6b8ea35da 100644 --- a/src/Grpc.Net.Client/Internal/StreamExtensions.cs +++ b/src/Grpc.Net.Client/Internal/StreamExtensions.cs @@ -22,9 +22,7 @@ using System.Diagnostics.CodeAnalysis; using System.Runtime.InteropServices; using Grpc.Core; -using Grpc.Net.Compression; using Grpc.Shared; -using Microsoft.Extensions.Logging; namespace Grpc.Net.Client.Internal; @@ -131,7 +129,7 @@ private static Status CreateUnknownMessageEncodingMessageStatus(string unsupport } // Performance improvement would be to decompress without converting to an intermediary byte array - if (!TryDecompressMessage(call.Logger, grpcEncoding, call.Channel.CompressionProviders, buffer, length, out var decompressedMessage)) + if (!TryDecompressMessage(call, grpcEncoding, buffer, length, out var decompressedMessage)) { var supportedEncodings = new List(); supportedEncodings.Add(GrpcProtocolConstants.IdentityGrpcEncoding); @@ -251,19 +249,20 @@ private static async Task ReadMessageContentAsync(Stream responseStream, Memory< } } - private static bool TryDecompressMessage(ILogger logger, string compressionEncoding, Dictionary compressionProviders, byte[] messageData, int length, out ReadOnlySequence result) + private static bool TryDecompressMessage(GrpcCall call, string compressionEncoding, byte[] messageData, int length, out ReadOnlySequence result) { - if (compressionProviders.TryGetValue(compressionEncoding, out var compressionProvider)) + if (call.Channel.CompressionProviders.TryGetValue(compressionEncoding, out var compressionProvider)) { - GrpcCallLog.DecompressingMessage(logger, compressionProvider.EncodingName); + GrpcCallLog.DecompressingMessage(call.Logger, compressionProvider.EncodingName); - var output = new MemoryStream(); using (var compressionStream = compressionProvider.CreateDecompressionStream(new MemoryStream(messageData, 0, length, writable: true, publiclyVisible: true))) { - compressionStream.CopyTo(output); + if (!CommonGrpcProtocolHelpers.TryReadCompressedMessage(compressionStream, call.Channel.ReceiveMaxMessageSize, out result)) + { + throw call.CreateRpcException(ReceivedMessageExceedsLimitStatus); + } } - result = new ReadOnlySequence(output.GetBuffer(), 0, (int)output.Length); return true; } diff --git a/src/Shared/CommonGrpcProtocolHelpers.cs b/src/Shared/CommonGrpcProtocolHelpers.cs index 78b824b4e..2914e5f30 100644 --- a/src/Shared/CommonGrpcProtocolHelpers.cs +++ b/src/Shared/CommonGrpcProtocolHelpers.cs @@ -16,6 +16,7 @@ #endregion +using System.Buffers; using System.Text; namespace Grpc.Shared; @@ -25,6 +26,41 @@ internal static class CommonGrpcProtocolHelpers public static readonly Task TrueTask = Task.FromResult(true); public static readonly Task FalseTask = Task.FromResult(false); + public static bool TryReadCompressedMessage(Stream compressionStream, int? maxReceiveMessageSize, out ReadOnlySequence message) + { + using var output = new MemoryStream(); + if (maxReceiveMessageSize == null) + { + compressionStream.CopyTo(output); + } + else + { + var buffer = ArrayPool.Shared.Rent(8192); + try + { + int read; + // Read at most one byte beyond the limit to detect oversized messages without fully decompressing them. + while ((read = compressionStream.Read(buffer, 0, (int)Math.Min(buffer.Length, maxReceiveMessageSize.Value - output.Length + 1))) > 0) + { + if (output.Length + read > maxReceiveMessageSize.Value) + { + message = default; + return false; + } + + output.Write(buffer, 0, read); + } + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + + message = new ReadOnlySequence(output.GetBuffer(), 0, (int)output.Length); + return true; + } + // Timer and DateTime.UtcNow have a 14ms precision. Add a small delay when scheduling deadline // timer that tests if exceeded or not. This avoids rescheduling the deadline callback multiple // times when timer is triggered before DateTime.UtcNow reports the deadline has been exceeded. diff --git a/test/FunctionalTests/Server/MaxMessageSizeTests.cs b/test/FunctionalTests/Server/MaxMessageSizeTests.cs index 043940267..6235d2c6e 100644 --- a/test/FunctionalTests/Server/MaxMessageSizeTests.cs +++ b/test/FunctionalTests/Server/MaxMessageSizeTests.cs @@ -18,6 +18,7 @@ using Greet; using Grpc.AspNetCore.FunctionalTests.Infrastructure; +using Grpc.AspNetCore.Server.Internal; using Grpc.Core; using Grpc.Tests.Shared; using NUnit.Framework; @@ -27,6 +28,34 @@ namespace Grpc.AspNetCore.FunctionalTests.Server; [TestFixture] public class MaxMessageSizeTests : FunctionalTestBase { + [TestCase("gzip")] + [TestCase("deflate")] + public async Task ReceivedCompressedMessageExceedsSize_ThrowError(string encoding) + { + SetExpectedErrorsFilter(writeContext => + writeContext.EventId.Name == "ErrorReadingMessage" && + GetRpcExceptionDetail(writeContext.Exception) == "Received message exceeds the maximum configured message size."); + + var invoked = false; + var method = Fixture.DynamicGrpc.AddUnaryMethod((request, context) => + { + invoked = true; + return Task.FromResult(new HelloReply()); + }); + using var ms = new MemoryStream(); + MessageHelpers.WriteMessage(ms, new HelloRequest { Name = new string('!', 4 * 1024 * 1024 + 1) }, encoding); + Assert.Less(ms.Length, 64 * 1024); + using var request = GrpcHttpHelper.Create(method.FullName); + request.Headers.Add(GrpcProtocolConstants.MessageEncodingHeader, encoding); + request.Content = new GrpcStreamContent(ms); + + using var response = await Fixture.Client.SendAsync(request).DefaultTimeout(); + + response.AssertTrailerStatus(StatusCode.ResourceExhausted, "Received message exceeds the maximum configured message size."); + Assert.IsFalse(invoked); + AssertHasLogRpcConnectionError(StatusCode.ResourceExhausted, "Received message exceeds the maximum configured message size."); + } + [Test] public async Task ReceivedMessageExceedsSize_ThrowError() { diff --git a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs index d12ba5f3e..3ad585bc9 100644 --- a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs +++ b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs @@ -17,6 +17,7 @@ #endregion using System.Buffers; +using System.Buffers.Binary; using System.IO.Pipelines; using System.Text; using Grpc.AspNetCore.Server.Internal; @@ -26,6 +27,7 @@ using Grpc.Net.Compression; using Grpc.Tests.Shared; using Microsoft.AspNetCore.Http; +using Moq; using NUnit.Framework; namespace Grpc.AspNetCore.Server.Tests; @@ -133,6 +135,191 @@ public async Task ReadSingleMessageAsync_ExceedReceiveSize_ReturnData() Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); } + [TestCase("gzip", false)] + [TestCase("gzip", true)] + [TestCase("deflate", false)] + [TestCase("deflate", true)] + public async Task ReadMessageAsync_CompressedMessage_EnforceReceiveSize(string encoding, bool streaming) + { + var provider = encoding == "gzip" + ? (ICompressionProvider)new GzipCompressionProvider(System.IO.Compression.CompressionLevel.Fastest) + : new DeflateCompressionProvider(System.IO.Compression.CompressionLevel.Fastest); + const int limit = 16 * 1024; + + foreach (var size in new[] { limit - 1, limit, limit + 1 }) + { + using var ms = CreateCompressedMessage(provider, new byte[size]); + Assert.Less(ms.Length - 5, limit); + var httpContext = new DefaultHttpContext(); + httpContext.Request.Headers[GrpcProtocolConstants.MessageEncodingHeader] = encoding; + var context = HttpContextServerCallContextHelper.CreateServerCallContext( + httpContext, + compressionProviders: new List { provider }, + maxReceiveMessageSize: limit); + var pipeReader = PipeReader.Create(ms); + var deserialized = false; + TestData Deserialize(DeserializationContext c) + { + deserialized = true; + return TestDataMarshaller.ContextualDeserializer(c); + } + + if (size > limit) + { + var ex = await ExceptionAssert.ThrowsAsync(async () => + { + if (streaming) + { + await pipeReader.ReadStreamMessageAsync(context, Deserialize); + } + else + { + await pipeReader.ReadSingleMessageAsync(context, Deserialize); + } + }).DefaultTimeout(); + + Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); + Assert.AreEqual("Received message exceeds the maximum configured message size.", ex.Status.Detail); + Assert.IsFalse(deserialized); + } + else + { + var message = streaming + ? await pipeReader.ReadStreamMessageAsync(context, Deserialize) + : await pipeReader.ReadSingleMessageAsync(context, Deserialize); + + Assert.IsTrue(deserialized); + Assert.AreEqual(size, message!.Span.Length); + } + } + } + + [TestCase(false, null)] + [TestCase(true, null)] + [TestCase(false, int.MaxValue)] + [TestCase(true, int.MaxValue)] + public async Task ReadMessageAsync_CompressedMessage_LargeOrUnlimitedReceiveSize_ReturnData(bool streaming, int? limit) + { + var provider = new GzipCompressionProvider(System.IO.Compression.CompressionLevel.Fastest); + var content = new byte[32 * 1024]; + using var ms = CreateCompressedMessage(provider, content); + var httpContext = new DefaultHttpContext(); + httpContext.Request.Headers[GrpcProtocolConstants.MessageEncodingHeader] = "gzip"; + var context = HttpContextServerCallContextHelper.CreateServerCallContext( + httpContext, + compressionProviders: new List { provider }, + maxReceiveMessageSize: limit); + var pipeReader = PipeReader.Create(ms); + + var message = streaming + ? await pipeReader.ReadStreamMessageAsync(context, TestDataMarshaller.ContextualDeserializer) + : await pipeReader.ReadSingleMessageAsync(context, TestDataMarshaller.ContextualDeserializer); + + CollectionAssert.AreEqual(content, message!.Span.ToArray()); + } + + [TestCase(false, 0)] + [TestCase(true, 0)] + [TestCase(false, 16 * 1024)] + [TestCase(true, 16 * 1024)] + public async Task ReadMessageAsync_CompressedMessage_StopsDecompressingAtReceiveSize(bool streaming, int limit) + { + using var decompressionStream = new TrackingDecompressionStream(new byte[128 * 1024]); + var provider = new Mock(); + provider.SetupGet(p => p.EncodingName).Returns("test"); + provider.Setup(p => p.CreateDecompressionStream(It.IsAny())).Returns(decompressionStream); + var httpContext = new DefaultHttpContext(); + httpContext.Request.Headers[GrpcProtocolConstants.MessageEncodingHeader] = "test"; + var context = HttpContextServerCallContextHelper.CreateServerCallContext( + httpContext, + compressionProviders: new List { provider.Object }, + maxReceiveMessageSize: limit); + using var ms = new MemoryStream(new byte[] { 1, 0, 0, 0, 0 }); + var pipeReader = PipeReader.Create(ms); + TestData Deserialize(DeserializationContext c) => throw new InvalidOperationException("Deserializer must not be called."); + + var ex = await ExceptionAssert.ThrowsAsync(async () => + { + if (streaming) + { + await pipeReader.ReadStreamMessageAsync(context, Deserialize); + } + else + { + await pipeReader.ReadSingleMessageAsync(context, Deserialize); + } + }).DefaultTimeout(); + + Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); + Assert.AreEqual(limit + 1, decompressionStream.BytesRead); + Assert.IsTrue(decompressionStream.Disposed); + } + + [TestCase(false)] + [TestCase(true)] + public async Task ReadMessageAsync_CompressedFrameExceedsReceiveSize_ThrowBeforeDecompressing(bool streaming) + { + var provider = new Mock(); + provider.SetupGet(p => p.EncodingName).Returns("test"); + var context = HttpContextServerCallContextHelper.CreateServerCallContext( + compressionProviders: new List { provider.Object }, + maxReceiveMessageSize: 1); + using var ms = new MemoryStream(new byte[] { 1, 0, 0, 0, 2, 0, 0 }); + var pipeReader = PipeReader.Create(ms); + + var ex = await ExceptionAssert.ThrowsAsync(async () => + { + if (streaming) + { + await pipeReader.ReadStreamMessageAsync(context, TestDataMarshaller.ContextualDeserializer); + } + else + { + await pipeReader.ReadSingleMessageAsync(context, TestDataMarshaller.ContextualDeserializer); + } + }).DefaultTimeout(); + + Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); + provider.Verify(p => p.CreateDecompressionStream(It.IsAny()), Times.Never); + } + + private static MemoryStream CreateCompressedMessage(ICompressionProvider provider, byte[] content) + { + var ms = new MemoryStream(); + ms.Write(new byte[] { 1, 0, 0, 0, 0 }); + using (var compressionStream = provider.CreateCompressionStream(ms, null)) + { + compressionStream.Write(content); + } + + BinaryPrimitives.WriteInt32BigEndian(ms.GetBuffer().AsSpan(1), (int)ms.Length - 5); + ms.Position = 0; + return ms; + } + + private sealed class TrackingDecompressionStream : MemoryStream + { + public int BytesRead { get; private set; } + public bool Disposed { get; private set; } + + public TrackingDecompressionStream(byte[] content) : base(content) + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + var read = base.Read(buffer, offset, count); + BytesRead += read; + return read; + } + + protected override void Dispose(bool disposing) + { + Disposed = true; + base.Dispose(disposing); + } + } + [Test] public async Task ReadSingleMessageAsync_LongMessage_ReturnData() { diff --git a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs index a930fc8b7..24a0bb2d7 100644 --- a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs +++ b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs @@ -21,7 +21,9 @@ using Grpc.Core; using Grpc.Net.Client.Internal; using Grpc.Net.Client.Tests.Infrastructure; +using Grpc.Net.Compression; using Grpc.Tests.Shared; +using Moq; using NUnit.Framework; namespace Grpc.Net.Client.Tests; @@ -29,6 +31,145 @@ namespace Grpc.Net.Client.Tests; [TestFixture] public class MaximumMessageSizeTests { + [TestCase("gzip", false)] + [TestCase("gzip", true)] +#if NET6_0_OR_GREATER + [TestCase("deflate", false)] + [TestCase("deflate", true)] +#endif + public async Task CompressedResponse_EnforceReceiveMaxMessageSize(string encoding, bool streaming) + { + var reply = new HelloReply { Message = new string('!', 16 * 1024) }; + var provider = GrpcProtocolConstants.DefaultCompressionProviders[encoding]; + + foreach (var limit in new int?[] { reply.CalculateSize() - 1, reply.CalculateSize(), reply.CalculateSize() + 1, null, int.MaxValue }) + { + using var httpClient = ClientTestHelpers.CreateTestClient(async request => + { + var content = await ClientTestHelpers.CreateResponseContent(reply, provider).DefaultTimeout(); + Assert.Less((await content.ReadAsByteArrayAsync()).Length - 5, reply.CalculateSize() - 1); + return ResponseUtils.CreateResponse(HttpStatusCode.OK, content, grpcEncoding: encoding); + }); + var invoker = HttpClientCallInvokerFactory.Create(httpClient, configure: o => o.MaxReceiveMessageSize = limit); + var deserialized = false; + var responseMarshaller = Marshallers.Create( + ClientTestHelpers.ServiceMethod.ResponseMarshaller.ContextualSerializer, + c => + { + deserialized = true; + return ClientTestHelpers.ServiceMethod.ResponseMarshaller.ContextualDeserializer(c); + }); + var method = new Method( + streaming ? MethodType.ServerStreaming : MethodType.Unary, + "ServiceName", + "MethodName", + ClientTestHelpers.ServiceMethod.RequestMarshaller, + responseMarshaller); + + async Task ReadResponseAsync() + { + if (streaming) + { + using var call = invoker.AsyncServerStreamingCall(method, string.Empty, new CallOptions(), new HelloRequest()); + Assert.IsTrue(await call.ResponseStream.MoveNext(CancellationToken.None).DefaultTimeout()); + var message = call.ResponseStream.Current; + Assert.IsFalse(await call.ResponseStream.MoveNext(CancellationToken.None).DefaultTimeout()); + return message; + } + else + { + using var call = invoker.AsyncUnaryCall(method, string.Empty, new CallOptions(), new HelloRequest()); + return await call.ResponseAsync.DefaultTimeout(); + } + } + + if (reply.CalculateSize() > limit) + { + var ex = await ExceptionAssert.ThrowsAsync(ReadResponseAsync).DefaultTimeout(); + Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); + Assert.AreEqual("Received message exceeds the maximum configured message size.", ex.Status.Detail); + Assert.IsFalse(deserialized); + } + else + { + var response = await ReadResponseAsync().DefaultTimeout(); + Assert.AreEqual(reply.Message, response.Message); + Assert.IsTrue(deserialized); + } + } + } + + [TestCase(false, 0)] + [TestCase(true, 0)] + [TestCase(false, 16 * 1024)] + [TestCase(true, 16 * 1024)] + public async Task CompressedResponse_StopsDecompressingAtReceiveMaxMessageSize(bool streaming, int limit) + { + using var decompressionStream = new TrackingDecompressionStream(new byte[128 * 1024]); + var provider = new Mock(); + provider.SetupGet(p => p.EncodingName).Returns("test"); + provider.Setup(p => p.CreateDecompressionStream(It.IsAny())).Returns(decompressionStream); + using var responseStream = new MemoryStream(new byte[] { 1, 0, 0, 0, 0 }); + + var ex = await ExceptionAssert.ThrowsAsync(() => StreamSerializationHelper.ReadMessageAsync( + responseStream, + c => throw new InvalidOperationException("Deserializer must not be called."), + "test", + limit, + new Dictionary { ["test"] = provider.Object }, + singleMessage: !streaming, + CancellationToken.None)).DefaultTimeout(); + + Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); + Assert.AreEqual("Received message exceeds the maximum configured message size.", ex.Status.Detail); + Assert.AreEqual(limit + 1, decompressionStream.BytesRead); + Assert.IsTrue(decompressionStream.Disposed); + } + + [TestCase(false)] + [TestCase(true)] + public async Task CompressedResponse_WireSizeExceedsReceiveMaxMessageSize_ThrowBeforeDecompressing(bool streaming) + { + var provider = new Mock(); + provider.SetupGet(p => p.EncodingName).Returns("test"); + using var responseStream = new MemoryStream(new byte[] { 1, 0, 0, 0, 2, 0, 0 }); + + var ex = await ExceptionAssert.ThrowsAsync(() => StreamSerializationHelper.ReadMessageAsync( + responseStream, + ClientTestHelpers.ServiceMethod.ResponseMarshaller.ContextualDeserializer, + "test", + 1, + new Dictionary { ["test"] = provider.Object }, + singleMessage: !streaming, + CancellationToken.None)).DefaultTimeout(); + + Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); + provider.Verify(p => p.CreateDecompressionStream(It.IsAny()), Times.Never); + } + + private sealed class TrackingDecompressionStream : MemoryStream + { + public int BytesRead { get; private set; } + public bool Disposed { get; private set; } + + public TrackingDecompressionStream(byte[] content) : base(content) + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + var read = base.Read(buffer, offset, count); + BytesRead += read; + return read; + } + + protected override void Dispose(bool disposing) + { + Disposed = true; + base.Dispose(disposing); + } + } + private async Task HandleRequest(HttpRequestMessage request) { var requestStream = await request.Content!.ReadAsStreamAsync().DefaultTimeout(); From 770150f82397cd8313f1b8093b24a3d9e41198d6 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Wed, 28 Oct 2026 15:49:00 +0800 Subject: [PATCH 2/4] Replace Moq in receive message size tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../PipeExtensionsTestsBase.cs | 38 ++++++++++++++----- .../MaximumMessageSizeTests.cs | 38 ++++++++++++++----- 2 files changed, 58 insertions(+), 18 deletions(-) diff --git a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs index 3ad585bc9..8e380b60f 100644 --- a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs +++ b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs @@ -27,7 +27,6 @@ using Grpc.Net.Compression; using Grpc.Tests.Shared; using Microsoft.AspNetCore.Http; -using Moq; using NUnit.Framework; namespace Grpc.AspNetCore.Server.Tests; @@ -225,14 +224,12 @@ public async Task ReadMessageAsync_CompressedMessage_LargeOrUnlimitedReceiveSize public async Task ReadMessageAsync_CompressedMessage_StopsDecompressingAtReceiveSize(bool streaming, int limit) { using var decompressionStream = new TrackingDecompressionStream(new byte[128 * 1024]); - var provider = new Mock(); - provider.SetupGet(p => p.EncodingName).Returns("test"); - provider.Setup(p => p.CreateDecompressionStream(It.IsAny())).Returns(decompressionStream); + var provider = new TestCompressionProvider(decompressionStream); var httpContext = new DefaultHttpContext(); httpContext.Request.Headers[GrpcProtocolConstants.MessageEncodingHeader] = "test"; var context = HttpContextServerCallContextHelper.CreateServerCallContext( httpContext, - compressionProviders: new List { provider.Object }, + compressionProviders: new List { provider }, maxReceiveMessageSize: limit); using var ms = new MemoryStream(new byte[] { 1, 0, 0, 0, 0 }); var pipeReader = PipeReader.Create(ms); @@ -259,10 +256,9 @@ public async Task ReadMessageAsync_CompressedMessage_StopsDecompressingAtReceive [TestCase(true)] public async Task ReadMessageAsync_CompressedFrameExceedsReceiveSize_ThrowBeforeDecompressing(bool streaming) { - var provider = new Mock(); - provider.SetupGet(p => p.EncodingName).Returns("test"); + var provider = new TestCompressionProvider(null); var context = HttpContextServerCallContextHelper.CreateServerCallContext( - compressionProviders: new List { provider.Object }, + compressionProviders: new List { provider }, maxReceiveMessageSize: 1); using var ms = new MemoryStream(new byte[] { 1, 0, 0, 0, 2, 0, 0 }); var pipeReader = PipeReader.Create(ms); @@ -280,7 +276,31 @@ public async Task ReadMessageAsync_CompressedFrameExceedsReceiveSize_ThrowBefore }).DefaultTimeout(); Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); - provider.Verify(p => p.CreateDecompressionStream(It.IsAny()), Times.Never); + Assert.IsFalse(provider.DecompressionStreamCreated); + } + + private sealed class TestCompressionProvider : ICompressionProvider + { + private readonly Stream? _decompressionStream; + + public TestCompressionProvider(Stream? decompressionStream) + { + _decompressionStream = decompressionStream; + } + + public string EncodingName => "test"; + public bool DecompressionStreamCreated { get; private set; } + + public Stream CreateCompressionStream(Stream stream, System.IO.Compression.CompressionLevel? compressionLevel) + { + throw new NotImplementedException(); + } + + public Stream CreateDecompressionStream(Stream stream) + { + DecompressionStreamCreated = true; + return _decompressionStream ?? throw new InvalidOperationException("Decompressor must not be called."); + } } private static MemoryStream CreateCompressedMessage(ICompressionProvider provider, byte[] content) diff --git a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs index 24a0bb2d7..299fc675a 100644 --- a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs +++ b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs @@ -23,7 +23,6 @@ using Grpc.Net.Client.Tests.Infrastructure; using Grpc.Net.Compression; using Grpc.Tests.Shared; -using Moq; using NUnit.Framework; namespace Grpc.Net.Client.Tests; @@ -106,9 +105,7 @@ async Task ReadResponseAsync() public async Task CompressedResponse_StopsDecompressingAtReceiveMaxMessageSize(bool streaming, int limit) { using var decompressionStream = new TrackingDecompressionStream(new byte[128 * 1024]); - var provider = new Mock(); - provider.SetupGet(p => p.EncodingName).Returns("test"); - provider.Setup(p => p.CreateDecompressionStream(It.IsAny())).Returns(decompressionStream); + var provider = new TestCompressionProvider(decompressionStream); using var responseStream = new MemoryStream(new byte[] { 1, 0, 0, 0, 0 }); var ex = await ExceptionAssert.ThrowsAsync(() => StreamSerializationHelper.ReadMessageAsync( @@ -116,7 +113,7 @@ public async Task CompressedResponse_StopsDecompressingAtReceiveMaxMessageSize(b c => throw new InvalidOperationException("Deserializer must not be called."), "test", limit, - new Dictionary { ["test"] = provider.Object }, + new Dictionary { ["test"] = provider }, singleMessage: !streaming, CancellationToken.None)).DefaultTimeout(); @@ -130,8 +127,7 @@ public async Task CompressedResponse_StopsDecompressingAtReceiveMaxMessageSize(b [TestCase(true)] public async Task CompressedResponse_WireSizeExceedsReceiveMaxMessageSize_ThrowBeforeDecompressing(bool streaming) { - var provider = new Mock(); - provider.SetupGet(p => p.EncodingName).Returns("test"); + var provider = new TestCompressionProvider(null); using var responseStream = new MemoryStream(new byte[] { 1, 0, 0, 0, 2, 0, 0 }); var ex = await ExceptionAssert.ThrowsAsync(() => StreamSerializationHelper.ReadMessageAsync( @@ -139,12 +135,36 @@ public async Task CompressedResponse_WireSizeExceedsReceiveMaxMessageSize_ThrowB ClientTestHelpers.ServiceMethod.ResponseMarshaller.ContextualDeserializer, "test", 1, - new Dictionary { ["test"] = provider.Object }, + new Dictionary { ["test"] = provider }, singleMessage: !streaming, CancellationToken.None)).DefaultTimeout(); Assert.AreEqual(StatusCode.ResourceExhausted, ex.StatusCode); - provider.Verify(p => p.CreateDecompressionStream(It.IsAny()), Times.Never); + Assert.IsFalse(provider.DecompressionStreamCreated); + } + + private sealed class TestCompressionProvider : ICompressionProvider + { + private readonly Stream? _decompressionStream; + + public TestCompressionProvider(Stream? decompressionStream) + { + _decompressionStream = decompressionStream; + } + + public string EncodingName => "test"; + public bool DecompressionStreamCreated { get; private set; } + + public Stream CreateCompressionStream(Stream stream, System.IO.Compression.CompressionLevel? compressionLevel) + { + throw new NotImplementedException(); + } + + public Stream CreateDecompressionStream(Stream stream) + { + DecompressionStreamCreated = true; + return _decompressionStream ?? throw new InvalidOperationException("Decompressor must not be called."); + } } private sealed class TrackingDecompressionStream : MemoryStream From 5cb2a71ac9b5d2fdf7331cd22d8be2d536ff0c88 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Wed, 28 Oct 2026 15:50:24 +0800 Subject: [PATCH 3/4] Share compression provider across receive size tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Grpc.AspNetCore.Server.Tests.csproj | 1 + .../PipeExtensionsTestsBase.cs | 24 ---------- .../Grpc.Net.Client.Tests.csproj | 1 + .../MaximumMessageSizeTests.cs | 24 ---------- test/Shared/TestCompressionProvider.cs | 45 +++++++++++++++++++ 5 files changed, 47 insertions(+), 48 deletions(-) create mode 100644 test/Shared/TestCompressionProvider.cs diff --git a/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj b/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj index ab2bed89b..e791885e2 100644 --- a/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj +++ b/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj @@ -22,6 +22,7 @@ + diff --git a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs index 8e380b60f..c06315736 100644 --- a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs +++ b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs @@ -279,30 +279,6 @@ public async Task ReadMessageAsync_CompressedFrameExceedsReceiveSize_ThrowBefore Assert.IsFalse(provider.DecompressionStreamCreated); } - private sealed class TestCompressionProvider : ICompressionProvider - { - private readonly Stream? _decompressionStream; - - public TestCompressionProvider(Stream? decompressionStream) - { - _decompressionStream = decompressionStream; - } - - public string EncodingName => "test"; - public bool DecompressionStreamCreated { get; private set; } - - public Stream CreateCompressionStream(Stream stream, System.IO.Compression.CompressionLevel? compressionLevel) - { - throw new NotImplementedException(); - } - - public Stream CreateDecompressionStream(Stream stream) - { - DecompressionStreamCreated = true; - return _decompressionStream ?? throw new InvalidOperationException("Decompressor must not be called."); - } - } - private static MemoryStream CreateCompressedMessage(ICompressionProvider provider, byte[] content) { var ms = new MemoryStream(); diff --git a/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj b/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj index f141c93e2..fd29c99c0 100644 --- a/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj +++ b/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj @@ -26,6 +26,7 @@ + diff --git a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs index 299fc675a..b6187596d 100644 --- a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs +++ b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs @@ -143,30 +143,6 @@ public async Task CompressedResponse_WireSizeExceedsReceiveMaxMessageSize_ThrowB Assert.IsFalse(provider.DecompressionStreamCreated); } - private sealed class TestCompressionProvider : ICompressionProvider - { - private readonly Stream? _decompressionStream; - - public TestCompressionProvider(Stream? decompressionStream) - { - _decompressionStream = decompressionStream; - } - - public string EncodingName => "test"; - public bool DecompressionStreamCreated { get; private set; } - - public Stream CreateCompressionStream(Stream stream, System.IO.Compression.CompressionLevel? compressionLevel) - { - throw new NotImplementedException(); - } - - public Stream CreateDecompressionStream(Stream stream) - { - DecompressionStreamCreated = true; - return _decompressionStream ?? throw new InvalidOperationException("Decompressor must not be called."); - } - } - private sealed class TrackingDecompressionStream : MemoryStream { public int BytesRead { get; private set; } diff --git a/test/Shared/TestCompressionProvider.cs b/test/Shared/TestCompressionProvider.cs new file mode 100644 index 000000000..714a6d86e --- /dev/null +++ b/test/Shared/TestCompressionProvider.cs @@ -0,0 +1,45 @@ +#region Copyright notice and license + +// Copyright 2026 The gRPC 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 +// +// http://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. + +#endregion + +using Grpc.Net.Compression; + +namespace Grpc.Tests.Shared; + +internal sealed class TestCompressionProvider : ICompressionProvider +{ + private readonly Stream? _decompressionStream; + + public TestCompressionProvider(Stream? decompressionStream) + { + _decompressionStream = decompressionStream; + } + + public string EncodingName => "test"; + public bool DecompressionStreamCreated { get; private set; } + + public Stream CreateCompressionStream(Stream stream, System.IO.Compression.CompressionLevel? compressionLevel) + { + throw new NotImplementedException(); + } + + public Stream CreateDecompressionStream(Stream stream) + { + DecompressionStreamCreated = true; + return _decompressionStream ?? throw new InvalidOperationException("Decompressor must not be called."); + } +} From ced209acf97217c2761885c7dce2cab4bdf8b939 Mon Sep 17 00:00:00 2001 From: James Newton-King Date: Wed, 28 Oct 2026 15:52:51 +0800 Subject: [PATCH 4/4] Share tracking decompression stream across receive size tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Grpc.AspNetCore.Server.Tests.csproj | 1 + .../PipeExtensionsTestsBase.cs | 23 ---------- .../Grpc.Net.Client.Tests.csproj | 1 + .../MaximumMessageSizeTests.cs | 23 ---------- test/Shared/TrackingDecompressionStream.cs | 42 +++++++++++++++++++ 5 files changed, 44 insertions(+), 46 deletions(-) create mode 100644 test/Shared/TrackingDecompressionStream.cs diff --git a/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj b/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj index e791885e2..1b58b0f3f 100644 --- a/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj +++ b/test/Grpc.AspNetCore.Server.Tests/Grpc.AspNetCore.Server.Tests.csproj @@ -23,6 +23,7 @@ + diff --git a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs index c06315736..f147a5b9b 100644 --- a/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs +++ b/test/Grpc.AspNetCore.Server.Tests/PipeExtensionsTestsBase.cs @@ -293,29 +293,6 @@ private static MemoryStream CreateCompressedMessage(ICompressionProvider provide return ms; } - private sealed class TrackingDecompressionStream : MemoryStream - { - public int BytesRead { get; private set; } - public bool Disposed { get; private set; } - - public TrackingDecompressionStream(byte[] content) : base(content) - { - } - - public override int Read(byte[] buffer, int offset, int count) - { - var read = base.Read(buffer, offset, count); - BytesRead += read; - return read; - } - - protected override void Dispose(bool disposing) - { - Disposed = true; - base.Dispose(disposing); - } - } - [Test] public async Task ReadSingleMessageAsync_LongMessage_ReturnData() { diff --git a/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj b/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj index fd29c99c0..f160a7382 100644 --- a/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj +++ b/test/Grpc.Net.Client.Tests/Grpc.Net.Client.Tests.csproj @@ -27,6 +27,7 @@ + diff --git a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs index b6187596d..5d1555c0e 100644 --- a/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs +++ b/test/Grpc.Net.Client.Tests/MaximumMessageSizeTests.cs @@ -143,29 +143,6 @@ public async Task CompressedResponse_WireSizeExceedsReceiveMaxMessageSize_ThrowB Assert.IsFalse(provider.DecompressionStreamCreated); } - private sealed class TrackingDecompressionStream : MemoryStream - { - public int BytesRead { get; private set; } - public bool Disposed { get; private set; } - - public TrackingDecompressionStream(byte[] content) : base(content) - { - } - - public override int Read(byte[] buffer, int offset, int count) - { - var read = base.Read(buffer, offset, count); - BytesRead += read; - return read; - } - - protected override void Dispose(bool disposing) - { - Disposed = true; - base.Dispose(disposing); - } - } - private async Task HandleRequest(HttpRequestMessage request) { var requestStream = await request.Content!.ReadAsStreamAsync().DefaultTimeout(); diff --git a/test/Shared/TrackingDecompressionStream.cs b/test/Shared/TrackingDecompressionStream.cs new file mode 100644 index 000000000..141878386 --- /dev/null +++ b/test/Shared/TrackingDecompressionStream.cs @@ -0,0 +1,42 @@ +#region Copyright notice and license + +// Copyright 2026 The gRPC 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 +// +// http://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. + +#endregion + +namespace Grpc.Tests.Shared; + +internal sealed class TrackingDecompressionStream : MemoryStream +{ + public int BytesRead { get; private set; } + public bool Disposed { get; private set; } + + public TrackingDecompressionStream(byte[] content) : base(content) + { + } + + public override int Read(byte[] buffer, int offset, int count) + { + var read = base.Read(buffer, offset, count); + BytesRead += read; + return read; + } + + protected override void Dispose(bool disposing) + { + Disposed = true; + base.Dispose(disposing); + } +}