From 3d3571aca3b7c5f56787c0c761ac83c77f917faa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ate=C5=9F=20G=C3=B6ral?= Date: Mon, 5 Oct 2026 16:59:30 -0400 Subject: [PATCH] Add an optional maximum size for received gRPC messages. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Ateş Göral --- context/getting-started.md | 22 +++ guides/getting-started/readme.md | 22 +++ lib/protocol/grpc/body/readable.rb | 68 ++++++-- test/protocol/grpc/body/maximum_size.rb | 214 ++++++++++++++++++++++++ 4 files changed, 316 insertions(+), 10 deletions(-) create mode 100644 test/protocol/grpc/body/maximum_size.rb diff --git a/context/getting-started.md b/context/getting-started.md index 5d905bb..5783613 100644 --- a/context/getting-started.md +++ b/context/getting-started.md @@ -108,6 +108,28 @@ end Protocol::GRPC::Status::NAMES[status] # => "OK" ``` +### Limiting Message Size + +To bound the size of messages received from a peer, pass `maximum_size` to the readable body: + +``` ruby +readable_body = Protocol::GRPC::Body::Readable.new( + response.body, + message_class: Hello::HelloReply, + maximum_size: 4 * 1024 * 1024 +) + +begin + message = readable_body.read +ensure + readable_body.close +end +``` + +The limit applies to each message's payload, excluding the five-byte frame prefix. Exactly the configured number of bytes is accepted. The declared frame length is checked before reading its payload, and gzip/deflate output is checked incrementally before it is accumulated. Exceeding either limit raises {ruby Protocol::GRPC::Error} with `RESOURCE_EXHAUSTED` status. + +The underlying HTTP body may already have buffered a chunk containing multiple frames; this option bounds gRPC message processing, not transport buffering. The default is `nil` (unlimited). A limit of zero allows only empty, uncompressed payloads. + ### Server Middleware Create a server middleware by subclassing `Protocol::GRPC::Middleware`: diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md index 5d905bb..5783613 100644 --- a/guides/getting-started/readme.md +++ b/guides/getting-started/readme.md @@ -108,6 +108,28 @@ end Protocol::GRPC::Status::NAMES[status] # => "OK" ``` +### Limiting Message Size + +To bound the size of messages received from a peer, pass `maximum_size` to the readable body: + +``` ruby +readable_body = Protocol::GRPC::Body::Readable.new( + response.body, + message_class: Hello::HelloReply, + maximum_size: 4 * 1024 * 1024 +) + +begin + message = readable_body.read +ensure + readable_body.close +end +``` + +The limit applies to each message's payload, excluding the five-byte frame prefix. Exactly the configured number of bytes is accepted. The declared frame length is checked before reading its payload, and gzip/deflate output is checked incrementally before it is accumulated. Exceeding either limit raises {ruby Protocol::GRPC::Error} with `RESOURCE_EXHAUSTED` status. + +The underlying HTTP body may already have buffered a chunk containing multiple frames; this option bounds gRPC message processing, not transport buffering. The default is `nil` (unlimited). A limit of zero allows only empty, uncompressed payloads. + ### Server Middleware Create a server middleware by subclassing `Protocol::GRPC::Middleware`: diff --git a/lib/protocol/grpc/body/readable.rb b/lib/protocol/grpc/body/readable.rb index b21ed4a..85e6dfb 100644 --- a/lib/protocol/grpc/body/readable.rb +++ b/lib/protocol/grpc/body/readable.rb @@ -36,8 +36,15 @@ def self.wrap(message, **options) # @parameter message_class [Class | Nil] Protobuf message class with .decode method. # If `nil`, returns raw binary data (useful for channel adapters) # @parameter encoding [String | Nil] Compression encoding (from grpc-encoding header) - def initialize(body, message_class: nil, encoding: nil) + # @parameter maximum_size [Integer | Nil] Maximum wire and decompressed payload size in bytes, excluding the frame prefix. `nil` disables the limit. + # @raises [ArgumentError] If the maximum size is not a non-negative integer or `nil`. + def initialize(body, message_class: nil, encoding: nil, maximum_size: nil) + unless maximum_size.nil? || (maximum_size.is_a?(Integer) && maximum_size >= 0) + raise ArgumentError, "Maximum size must be a non-negative integer or nil!" + end + super(body) + @maximum_size = maximum_size @message_class = message_class @encoding = encoding @buffer = String.new.force_encoding(Encoding::BINARY) @@ -46,10 +53,14 @@ def initialize(body, message_class: nil, encoding: nil) # @attribute [String | Nil] The compression encoding. attr_reader :encoding + # @attribute [Integer | Nil] The maximum wire and decompressed payload size in bytes. + attr_reader :maximum_size + # Read the next gRPC message. # Overrides Wrapper#read to transform raw HTTP body chunks into decoded gRPC messages. # Errors raised by the underlying body propagate unchanged. # @returns [Object | String | Nil] Decoded message, raw binary, or `Nil` if stream ended + # @raises [Error] With `RESOURCE_EXHAUSTED` if a message exceeds the maximum size. def read # Read 5-byte prefix: 1 byte compression flag + 4 bytes length prefix = read_exactly(5) @@ -57,6 +68,7 @@ def read compressed = prefix[0].unpack1("C") == 1 length = prefix[1..4].unpack1("N") + check_size!(length) # Read the message body: data = read_exactly(length) @@ -106,26 +118,62 @@ def read_exactly(n) end # Decompress data using the configured encoding. - # @parameter data [String] The compressed data - # @returns [String] The decompressed data - # @raises [Error] If decompression fails + # @parameter data [String] The compressed data. + # @returns [String] The decompressed data. + # @raises [Error] If decompression fails or the maximum size is exceeded. def decompress(data) case @encoding when "gzip" - begin + if @maximum_size + inflate(data, Zlib::MAX_WBITS + 16) + else Zlib.gunzip(data) - rescue => error - raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}") end when "deflate" - begin + if @maximum_size + inflate(data, Zlib::MAX_WBITS) + else Zlib::Inflate.inflate(data) - rescue => error - raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}") end else raise Error.new(Status::UNIMPLEMENTED, "Unsupported compression encoding: #{@encoding.inspect}") end + rescue Zlib::Error => error + raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}") + end + + # Inflate a message without accumulating output beyond the maximum size. + # @parameter data [String] The compressed data. + # @parameter window_bits [Integer] The compression format. + # @returns [String] The decompressed data. + # @raises [Error] If the message is truncated or the maximum size is exceeded. + # @raises [Zlib::Error] If decompression fails. + def inflate(data, window_bits) + inflater = Zlib::Inflate.new(window_bits) + result = String.new.force_encoding(Encoding::BINARY) + + # Check each output chunk before retaining it: + inflater.inflate(data) do |chunk| + check_size!(result.bytesize + chunk.bytesize) + result << chunk + end + + unless inflater.finished? + raise Error.new(Status::INTERNAL, "Truncated compressed gRPC message!") + end + + result + ensure + inflater&.close + end + + # Check a message size against the configured maximum. + # @parameter size [Integer] The payload size in bytes. + # @raises [Error] With `RESOURCE_EXHAUSTED` if the maximum size is exceeded. + def check_size!(size) + if @maximum_size && size > @maximum_size + raise Error.new(Status::RESOURCE_EXHAUSTED, "Received message larger than maximum size: #{size} > #{@maximum_size}") + end end end end diff --git a/test/protocol/grpc/body/maximum_size.rb b/test/protocol/grpc/body/maximum_size.rb new file mode 100644 index 0000000..a7625e5 --- /dev/null +++ b/test/protocol/grpc/body/maximum_size.rb @@ -0,0 +1,214 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Ateş Göral. + +require "protocol/grpc/body/readable" +require "protocol/http/body/buffered" +require "protocol/http/body/writable" +require_relative "../../../../fixtures/protocol/grpc/test_message" + +require "zlib" + +describe Protocol::GRPC::Body::Readable do + with "maximum size" do + let(:source_body) {Protocol::HTTP::Body::Buffered.new} + let(:body) {subject.new(source_body, maximum_size: 1024)} + + def frame(data, compressed: false) + [compressed ? 1 : 0, data.bytesize].pack("CN") + data + end + + def expect_size_error + expect{yield}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED + expect(error.message).to be =~ /Received message larger than maximum size: \d+ > 1024/ + end + end + + it "defaults to unlimited messages" do + data = "a" * 2048 + source_body.write(frame(data)) + reader = subject.new(source_body) + + expect(reader.maximum_size).to be_nil + expect(reader.read).to be == data + end + + [-1, 1.5, "1024", false].each do |maximum_size| + it "rejects an invalid maximum size #{maximum_size.inspect}" do + expect{subject.new(source_body, maximum_size: maximum_size)}.to raise_exception(ArgumentError) + end + end + + it "accepts exactly the maximum payload size, excluding the prefix" do + data = "a" * 1024 + source_body.write(frame(data)) + + expect(body.maximum_size).to be == 1024 + expect(body.read).to be == data + end + + it "accepts empty messages with a zero limit" do + source_body.write(frame("")) + reader = subject.new(source_body, maximum_size: 0) + + expect(reader.read).to be == "" + expect(reader.read).to be_nil + end + + it "rejects non-empty messages with a zero limit" do + source_body.write(frame("a")) + reader = subject.new(source_body, maximum_size: 0) + + expect{reader.read}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED + end + end + + it "rejects an oversized declared payload without reading it" do + source_body.write([0, 1025].pack("CN")) + reads = 0 + mock(source_body) do |wrapper| + wrapper.wrap(:read) do |original| + reads += 1 + raise "Payload must not be read!" if reads > 1 + original.call + end + end + + expect_size_error {body.read} + expect(reads).to be == 1 + end + + it "reads fragmented prefixes and payloads" do + data = "a" * 1024 + frame(data).bytes.each_slice(3) do |bytes| + source_body.write(bytes.pack("C*")) + end + + expect(body.read).to be == data + expect(body.read).to be_nil + end + + it "limits each coalesced message independently" do + data = "a" * 1024 + source_body.write(frame(data) + frame(data)) + + expect(body.read).to be == data + expect(source_body).to be(:empty?) + expect(body.read).to be == data + expect(body.read).to be_nil + end + + it "rejects an oversized message after a complete coalesced message" do + source_body.write(frame("a") + frame("b" * 1025)) + + expect(body.read).to be == "a" + expect_size_error {body.read} + end + + it "preserves protobuf decoding within the limit" do + message_class = Protocol::GRPC::Fixtures::TestMessage + message = message_class.new(value: "Hello") + source_body.write(frame(message.to_proto)) + reader = subject.new(source_body, message_class: message_class, maximum_size: message.to_proto.bytesize) + + expect(reader.read).to be == message + end + + ["\x00\x00".b, [0, 2].pack("CN"), [0, 2].pack("CN") + "a"].each do |partial_frame| + it "rejects a truncated #{partial_frame.bytesize}-byte frame" do + source_body.write(partial_frame) + + expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL + end + end + end + + it "rejects an incomplete next frame after a complete message" do + source_body.write(frame("a") + "\x00\x00".b) + + expect(body.read).to be == "a" + expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL + end + end + + it "preserves a pending underlying read error" do + source = Protocol::HTTP::Body::Writable.new + error = IOError.new("Stream failed!") + source.close_write(error) + reader = subject.new(source, maximum_size: 1024) + + expect{reader.read}.to raise_exception(IOError).and(be_equal(error)) + end + + {"gzip" => Zlib.method(:gzip), "deflate" => Zlib::Deflate.method(:deflate)}.each do |encoding, compress| + with encoding do + let(:body) {subject.new(source_body, encoding: encoding, maximum_size: 1024)} + + it "accepts exactly the maximum decompressed size" do + data = "a" * 1024 + source_body.write(frame(compress.call(data), compressed: true)) + + expect(body.read).to be == data + end + + it "rejects expansion beyond the limit before accumulating the full output" do + data = "a" * 1_000_000 + compressed = compress.call(data) + expect(compressed.bytesize).to be < 1024 + source_body.write(frame(compressed, compressed: true)) + sizes = [] + mock(body) do |wrapper| + wrapper.wrap(:check_size!) do |original, size| + sizes << size + original.call(size) + end + end + + expect_size_error {body.read} + expect(sizes.last).to be > 1024 + expect(sizes.last).to be < data.bytesize + end + + it "resets the expanded-size limit for each message" do + data = "a" * 1024 + message = frame(compress.call(data), compressed: true) + source_body.write(message + message) + + expect(body.read).to be == data + expect(body.read).to be == data + end + + it "rejects a truncated compressed stream" do + compressed = compress.call("a" * 512).byteslice(0...-1) + source_body.write(frame(compressed, compressed: true)) + + expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL + end + end + + it "rejects invalid compressed data" do + source_body.write(frame("invalid", compressed: true)) + + expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL + end + end + end + end + + it "preserves unsupported compression errors" do + source_body.write(frame("data", compressed: true)) + reader = subject.new(source_body, encoding: "custom", maximum_size: 1024) + + expect{reader.read}.to raise_exception(Protocol::GRPC::Error) do |error| + expect(error.status_code).to be == Protocol::GRPC::Status::UNIMPLEMENTED + end + end + end +end