From 361e47d9b0b5948aa70ef3d83f57c340cd8a4571 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sat, 11 Jul 2026 22:27:56 +0200
Subject: [PATCH 01/18] AVRO-4297: [ruby] Validate available bytes before
 allocating for length-prefixed values and collections

A bytes or string value is a length prefix followed by that many bytes, and an
array or map block is an element count followed by that many items. A malicious
or truncated input can declare a huge length or count with little or no data.

- BinaryDecoder#bytes_remaining reports the bytes still readable when the reader
  can report its size (else nil). #read uses it to reject an over-large declared
  length above a threshold before allocating.
- DatumReader#read_array/#read_map reject a block whose element count could not
  be backed by the bytes remaining, using min_bytes_per_element computed from
  the element schema so a zero-byte element type (e.g. null) is not falsely
  rejected.

Mirrors the Java SDK's checks (AVRO-4241). Readers that cannot report their size
are unaffected.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb  | 67 ++++++++++++++++++++++++++++++++++++++-
 lang/ruby/test/test_io.rb | 64 +++++++++++++++++++++++++++++++++++++
 2 files changed, 130 insertions(+), 1 deletion(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 091b3544672..7d11b3ef5b8 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -37,6 +37,13 @@ def initialize(writers_schema, readers_schema)
     class BinaryDecoder
       # Read leaf values
 
+      # Reads with a declared length above this many bytes are validated
+      # against the number of bytes actually remaining (when the reader can
+      # report its size) before allocating, to guard against an out-of-memory
+      # attack from a malicious or truncated input. Smaller reads skip the
+      # check to avoid per-value overhead.
+      MAX_UNCHECKED_READ = 1024 * 1024
+
       # reader is an object on which we can call read, seek and tell.
       attr_reader :reader
       def initialize(reader)
@@ -103,10 +110,28 @@ def read_string
       end
 
       def read(len)
-        # Read n bytes
+        # Read n bytes. Reject a declared length that exceeds the bytes
+        # actually remaining before allocating for it, to guard against an
+        # out-of-memory attack from a malicious or truncated input. The check
+        # is only applied to larger reads; smaller reads and stream readers that
+        # cannot report their size fall back to reading directly.
+        if len > MAX_UNCHECKED_READ
+          remaining = bytes_remaining
+          if remaining && len > remaining
+            raise AvroError, "Cannot read #{len} bytes, only #{remaining} remaining"
+          end
+        end
         @reader.read(len)
       end
 
+      # Number of bytes still available to read, or nil when the reader cannot
+      # report its size. Used to reject a declared length or collection block
+      # count that exceeds the data actually available before allocating for it.
+      def bytes_remaining
+        return nil unless @reader.respond_to?(:size) && @reader.respond_to?(:tell)
+        @reader.size - @reader.tell
+      end
+
       def skip_null
         nil
       end
@@ -318,6 +343,7 @@ def read_array(writers_schema, readers_schema, decoder)
             block_count = -block_count
             _block_size = decoder.read_long
           end
+          ensure_collection_available(decoder, block_count, self.class.min_bytes_per_element(writers_schema.items))
           block_count.times do
             read_items << read_data(writers_schema.items,
                                     readers_schema.items,
@@ -337,6 +363,8 @@ def read_map(writers_schema, readers_schema, decoder)
             block_count = -block_count
             _block_size = decoder.read_long
           end
+          # Map keys are strings (>= 1 byte length prefix) plus the value.
+          ensure_collection_available(decoder, block_count, 1 + self.class.min_bytes_per_element(writers_schema.values))
           block_count.times do
             key = decoder.read_string
             read_items[key] = read_data(writers_schema.values,
@@ -496,6 +524,43 @@ def skip_record(writers_schema, decoder)
       end
 
       private
+      # Minimum number of bytes a single value of the given schema can occupy on
+      # the wire. Used to reject an array/map block count that could not be
+      # backed by the bytes remaining. A type that can encode to zero bytes
+      # (null) returns 0, which disables the check for it (so an array of nulls
+      # is not falsely rejected).
+      def self.min_bytes_per_element(schema, visited = nil)
+        visited ||= {}
+        case schema.type_sym
+        when :null then 0
+        when :float then 4
+        when :double then 8
+        when :fixed then schema.size
+        when :record, :error
+          return 0 if visited[schema.object_id]
+          visited[schema.object_id] = true
+          total = schema.fields.sum { |field| min_bytes_per_element(field.type, visited) }
+          visited.delete(schema.object_id)
+          total
+        else
+          # boolean, int, long, bytes, string, enum, union, array, map: >= 1 byte
+          # (a union encodes at least a 1-byte branch index).
+          1
+        end
+      end
+
+      # Reject a collection (array or map) block whose declared element count
+      # could not be backed by the bytes actually remaining, before iterating.
+      # Skipped when the per-element minimum is zero or when the reader cannot
+      # report how many bytes remain.
+      def ensure_collection_available(decoder, count, min_bytes_per_element)
+        return if count <= 0 || min_bytes_per_element <= 0
+        remaining = decoder.bytes_remaining
+        if remaining && count * min_bytes_per_element > remaining
+          raise AvroError, "Collection claims #{count} elements with at least #{min_bytes_per_element} bytes each, but only #{remaining} bytes are available"
+        end
+      end
+
       def skip_blocks(decoder, &blk)
         block_count = decoder.read_long
         while block_count != 0
diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index c2b5ffc722c..11dd08d3f8f 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -42,6 +42,70 @@ def test_bytes
     check_default('"bytes"', '"foo"', "foo")
   end
 
+  # A bytes/string value declares a length prefix; a malicious or truncated
+  # input can declare far more bytes than actually exist. On a reader that can
+  # report its size, that is rejected before allocating for it.
+  def test_read_bytes_rejects_length_beyond_stream
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(100 * 1024 * 1024)
+    reader = StringIO.new(writer.string)
+    decoder = Avro::IO::BinaryDecoder.new(reader)
+    assert_raise(Avro::AvroError) { decoder.read_bytes }
+  end
+
+  def test_read_bytes_within_stream_still_reads
+    payload = 'x' * (2 * 1024 * 1024)
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_bytes(payload)
+    reader = StringIO.new(writer.string)
+    decoder = Avro::IO::BinaryDecoder.new(reader)
+    assert_equal(payload, decoder.read_bytes)
+  end
+
+  # An array/map block declares an element count; a malicious or truncated input
+  # can declare far more elements than the remaining bytes could hold. The count
+  # is validated against the bytes remaining before iterating, using the minimum
+  # on-wire size of the element schema (so 0-byte elements like null are not
+  # falsely rejected).
+  def decode(schema_json, encoded)
+    schema = Avro::Schema.parse(schema_json)
+    reader = Avro::IO::DatumReader.new(schema)
+    reader.read(Avro::IO::BinaryDecoder.new(StringIO.new(encoded)))
+  end
+
+  def test_read_array_rejects_count_beyond_stream
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(1_000_000)
+    assert_raise(Avro::AvroError) do
+      decode('{"type":"array","items":"long"}', writer.string)
+    end
+  end
+
+  def test_read_map_rejects_count_beyond_stream
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(1_000_000)
+    assert_raise(Avro::AvroError) do
+      decode('{"type":"map","values":"long"}', writer.string)
+    end
+  end
+
+  def test_read_array_of_null_not_falsely_rejected
+    count = 100_000
+    writer = StringIO.new
+    encoder = Avro::IO::BinaryEncoder.new(writer)
+    encoder.write_long(count) # one block of `count` nulls (zero bytes each)
+    encoder.write_long(0)     # end-of-array marker
+    result = decode('{"type":"array","items":"null"}', writer.string)
+    assert_equal([nil] * count, result)
+  end
+
+  def test_read_array_within_stream_still_reads
+    schema = Avro::Schema.parse('{"type":"array","items":"long"}')
+    writer = StringIO.new
+    Avro::IO::DatumWriter.new(schema).write([1, 2, 3], Avro::IO::BinaryEncoder.new(writer))
+    assert_equal([1, 2, 3], decode('{"type":"array","items":"long"}', writer.string))
+  end
+
   def test_int
     check('"int"')
     check_default('"int"', "5", 5)

From ae87eb53e2f01e725838addbccea209411ef824e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sat, 11 Jul 2026 23:32:17 +0200
Subject: [PATCH 02/18] AVRO-4297: [ruby] Fix rubocop offenses in
 min_bytes_per_element

The Lint CI step (rubocop) flagged three offenses in the new code:
- Lint/IneffectiveAccessModifier: 'private' does not apply to a 'def self.'
  singleton method. Make min_bytes_per_element a private instance method (it
  uses no class state) and call it unqualified.
- Lint/HashCompareByIdentity: use a hash with compare_by_identity and the
  schema object as the key instead of keying on schema.object_id.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 7d11b3ef5b8..388f97b63db 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -343,7 +343,7 @@ def read_array(writers_schema, readers_schema, decoder)
             block_count = -block_count
             _block_size = decoder.read_long
           end
-          ensure_collection_available(decoder, block_count, self.class.min_bytes_per_element(writers_schema.items))
+          ensure_collection_available(decoder, block_count, min_bytes_per_element(writers_schema.items))
           block_count.times do
             read_items << read_data(writers_schema.items,
                                     readers_schema.items,
@@ -364,7 +364,7 @@ def read_map(writers_schema, readers_schema, decoder)
             _block_size = decoder.read_long
           end
           # Map keys are strings (>= 1 byte length prefix) plus the value.
-          ensure_collection_available(decoder, block_count, 1 + self.class.min_bytes_per_element(writers_schema.values))
+          ensure_collection_available(decoder, block_count, 1 + min_bytes_per_element(writers_schema.values))
           block_count.times do
             key = decoder.read_string
             read_items[key] = read_data(writers_schema.values,
@@ -529,18 +529,18 @@ def skip_record(writers_schema, decoder)
       # backed by the bytes remaining. A type that can encode to zero bytes
       # (null) returns 0, which disables the check for it (so an array of nulls
       # is not falsely rejected).
-      def self.min_bytes_per_element(schema, visited = nil)
-        visited ||= {}
+      def min_bytes_per_element(schema, visited = nil)
+        visited ||= {}.compare_by_identity
         case schema.type_sym
         when :null then 0
         when :float then 4
         when :double then 8
         when :fixed then schema.size
         when :record, :error
-          return 0 if visited[schema.object_id]
-          visited[schema.object_id] = true
+          return 0 if visited[schema]
+          visited[schema] = true
           total = schema.fields.sum { |field| min_bytes_per_element(field.type, visited) }
-          visited.delete(schema.object_id)
+          visited.delete(schema)
           total
         else
           # boolean, int, long, bytes, string, enum, union, array, map: >= 1 byte

From d4eec597a921b76de4336498d45a3b4c7cc17778 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 00:24:02 +0200
Subject: [PATCH 03/18] AVRO-4297: [ruby] Reject negative read lengths; treat
 request like record

Review feedback:
- BinaryDecoder#read now rejects a negative length. Ruby IO#read(-1) reads the
  rest of the stream, which would bypass the size check and allocate without
  bound.
- min_bytes_per_element treats :request like :record/:error (request schemas are
  record-like), so an array/map of request values whose fields encode to zero
  bytes is not falsely rejected.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 388f97b63db..baa6266788d 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -115,6 +115,11 @@ def read(len)
         # out-of-memory attack from a malicious or truncated input. The check
         # is only applied to larger reads; smaller reads and stream readers that
         # cannot report their size fall back to reading directly.
+        if len < 0
+          # A negative length would make IO#read return the rest of the stream,
+          # which bypasses the size check and can allocate without bound.
+          raise AvroError, "Cannot read a negative number of bytes: #{len}"
+        end
         if len > MAX_UNCHECKED_READ
           remaining = bytes_remaining
           if remaining && len > remaining
@@ -536,7 +541,7 @@ def min_bytes_per_element(schema, visited = nil)
         when :float then 4
         when :double then 8
         when :fixed then schema.size
-        when :record, :error
+        when :record, :error, :request
           return 0 if visited[schema]
           visited[schema] = true
           total = schema.fields.sum { |field| min_bytes_per_element(field.type, visited) }

From 34859b4a72bd279eddcfebd352f6d62ac91b9e7f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 01:02:58 +0200
Subject: [PATCH 04/18] AVRO-4297: [ruby] Make bytes_remaining resilient;
 compare via division

Review feedback:
- bytes_remaining now verifies #size/#tell return Integers and rescues
  IOError/SystemCallError/NotImplementedError, returning nil when the remaining
  size cannot be reported, so reads stay unaffected in that case.
- ensure_collection_available compares count against remaining /
  min_bytes_per_element rather than multiplying, avoiding a large intermediate
  Integer for an attacker-controlled count.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index baa6266788d..074ca7cd5ce 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -134,7 +134,14 @@ def read(len)
       # count that exceeds the data actually available before allocating for it.
       def bytes_remaining
         return nil unless @reader.respond_to?(:size) && @reader.respond_to?(:tell)
-        @reader.size - @reader.tell
+        size = @reader.size
+        pos = @reader.tell
+        return nil unless size.is_a?(Integer) && pos.is_a?(Integer)
+        size - pos
+      rescue IOError, SystemCallError, NotImplementedError
+        # The reader responds to #size/#tell but cannot actually report them;
+        # treat the remaining size as unknown and skip the check.
+        nil
       end
 
       def skip_null
@@ -561,7 +568,9 @@ def min_bytes_per_element(schema, visited = nil)
       def ensure_collection_available(decoder, count, min_bytes_per_element)
         return if count <= 0 || min_bytes_per_element <= 0
         remaining = decoder.bytes_remaining
-        if remaining && count * min_bytes_per_element > remaining
+        # Compare via integer division rather than multiplying, so a huge count
+        # does not create a large intermediate product.
+        if remaining && count > remaining / min_bytes_per_element
           raise AvroError, "Collection claims #{count} elements with at least #{min_bytes_per_element} bytes each, but only #{remaining} bytes are available"
         end
       end

From ea1ac32fbe8c9fe72e99f9f814ec8926eaca64fc Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 01:27:07 +0200
Subject: [PATCH 05/18] AVRO-4297: [ruby] Clamp negative bytes_remaining; cache
 collection minimum

Review feedback:
- bytes_remaining clamps a negative result (size smaller than the current
  position, e.g. a truncated file) to 0 with [size - pos, 0].max, so callers see
  no bytes available rather than a confusing negative count.
- read_array/read_map compute the per-element minimum once per invocation and
  reuse it across blocks, avoiding repeated schema traversal on the hot path.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 074ca7cd5ce..5da8b81f2dd 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -137,7 +137,10 @@ def bytes_remaining
         size = @reader.size
         pos = @reader.tell
         return nil unless size.is_a?(Integer) && pos.is_a?(Integer)
-        size - pos
+        # Clamp negative (e.g. the file was truncated below the current position)
+        # to 0 so callers see "no bytes available" rather than a confusing
+        # negative count.
+        [size - pos, 0].max
       rescue IOError, SystemCallError, NotImplementedError
         # The reader responds to #size/#tell but cannot actually report them;
         # treat the remaining size as unknown and skip the check.
@@ -349,13 +352,14 @@ def read_enum(writers_schema, readers_schema, decoder)
 
       def read_array(writers_schema, readers_schema, decoder)
         read_items = []
+        min_bytes = min_bytes_per_element(writers_schema.items)
         block_count = decoder.read_long
         while block_count != 0
           if block_count < 0
             block_count = -block_count
             _block_size = decoder.read_long
           end
-          ensure_collection_available(decoder, block_count, min_bytes_per_element(writers_schema.items))
+          ensure_collection_available(decoder, block_count, min_bytes)
           block_count.times do
             read_items << read_data(writers_schema.items,
                                     readers_schema.items,
@@ -369,14 +373,15 @@ def read_array(writers_schema, readers_schema, decoder)
 
       def read_map(writers_schema, readers_schema, decoder)
         read_items = {}
+        # Map keys are strings (>= 1 byte length prefix) plus the value.
+        min_bytes = 1 + min_bytes_per_element(writers_schema.values)
         block_count = decoder.read_long
         while block_count != 0
           if block_count < 0
             block_count = -block_count
             _block_size = decoder.read_long
           end
-          # Map keys are strings (>= 1 byte length prefix) plus the value.
-          ensure_collection_available(decoder, block_count, 1 + min_bytes_per_element(writers_schema.values))
+          ensure_collection_available(decoder, block_count, min_bytes)
           block_count.times do
             key = decoder.read_string
             read_items[key] = read_data(writers_schema.values,

From 1186853c6994120b9db657cc7d2b52ea5f6817c9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 07:20:09 +0200
Subject: [PATCH 06/18] AVRO-4297: [ruby] Skip the collection check for
 decoders without bytes_remaining

Review feedback: a decoder that implements the read protocol but not
#bytes_remaining (e.g. a custom decoder) would raise NoMethodError. Guard the
call with respond_to?(:bytes_remaining) and skip the check for such a decoder
instead of raising.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 5da8b81f2dd..5d739d9be96 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -572,6 +572,10 @@ def min_bytes_per_element(schema, visited = nil)
       # report how many bytes remain.
       def ensure_collection_available(decoder, count, min_bytes_per_element)
         return if count <= 0 || min_bytes_per_element <= 0
+        # A decoder that implements the read protocol but not #bytes_remaining
+        # (e.g. a custom decoder) cannot report the remaining size; skip the
+        # check for it rather than raising NoMethodError.
+        return unless decoder.respond_to?(:bytes_remaining)
         remaining = decoder.bytes_remaining
         # Compare via integer division rather than multiplying, so a huge count
         # does not create a large intermediate product.

From 1a3d3daf7740020f6ba844f46182cd2740737715 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 08:02:22 +0200
Subject: [PATCH 07/18] AVRO-4297: [ruby] Tie large-read tests to
 MAX_UNCHECKED_READ; lighter null assertion

Review feedback:
- Derive the oversized declared length and the within-limit payload from
  BinaryDecoder::MAX_UNCHECKED_READ + 1 instead of hard-coded 100MiB/2MiB
  constants, so the tests stay tied to the threshold under test and use less
  time and memory.
- Assert the decoded array length and spot-check a couple of nil elements
  instead of building a second large [nil] * count array.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/test/test_io.rb | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index 11dd08d3f8f..25314600006 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -47,14 +47,15 @@ def test_bytes
   # report its size, that is rejected before allocating for it.
   def test_read_bytes_rejects_length_beyond_stream
     writer = StringIO.new
-    Avro::IO::BinaryEncoder.new(writer).write_long(100 * 1024 * 1024)
+    declared = Avro::IO::BinaryDecoder::MAX_UNCHECKED_READ + 1
+    Avro::IO::BinaryEncoder.new(writer).write_long(declared)
     reader = StringIO.new(writer.string)
     decoder = Avro::IO::BinaryDecoder.new(reader)
     assert_raise(Avro::AvroError) { decoder.read_bytes }
   end
 
   def test_read_bytes_within_stream_still_reads
-    payload = 'x' * (2 * 1024 * 1024)
+    payload = 'x' * (Avro::IO::BinaryDecoder::MAX_UNCHECKED_READ + 1)
     writer = StringIO.new
     Avro::IO::BinaryEncoder.new(writer).write_bytes(payload)
     reader = StringIO.new(writer.string)
@@ -96,7 +97,9 @@ def test_read_array_of_null_not_falsely_rejected
     encoder.write_long(count) # one block of `count` nulls (zero bytes each)
     encoder.write_long(0)     # end-of-array marker
     result = decode('{"type":"array","items":"null"}', writer.string)
-    assert_equal([nil] * count, result)
+    assert_equal(count, result.length)
+    assert_nil(result.first)
+    assert_nil(result.last)
   end
 
   def test_read_array_within_stream_still_reads

From c6acf7722d7f626dd9a47a6fce96028f1fcdd350 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 16:49:02 +0200
Subject: [PATCH 08/18] AVRO-4297: [ruby] Cap zero-byte collection element
 allocation

Completes the available-bytes protection for collections. Elements whose schema
encodes to zero bytes (null, or a record with only zero-byte fields) consume no
input, so the bytes-remaining check cannot bound their count. A tiny payload
declaring a huge array block count of such elements (e.g.
{"type":"array","items":"null"} with a count of 200,000,000) therefore drove an
unbounded allocation.

ensure_collection_available now tracks the cumulative count across blocks and
enforces, per block: a structural cap on all collections
(DEFAULT_MAX_COLLECTION_STRUCTURAL = Integer.MAX_VALUE - 8, also covering
readers that cannot report their remaining bytes); a zero-byte item cap
(DEFAULT_MAX_COLLECTION_ITEMS = 10,000,000) when the per-element minimum is
zero; and the existing bytes-remaining check otherwise. AVRO_MAX_COLLECTION_ITEMS,
when set, caps both. read_array/read_map and the skip path (skip_blocks, used by
skip_array/skip_map) are all bounded the same way, so skipping a huge zero-byte
block cannot loop unboundedly. Rejections raise the dedicated
Avro::IO::CollectionSizeError (a subclass of AvroError).

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb  | 92 ++++++++++++++++++++++++++++++---------
 lang/ruby/test/test_io.rb | 42 +++++++++++++++++-
 2 files changed, 112 insertions(+), 22 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 5d739d9be96..eb632b24d4b 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -32,6 +32,10 @@ def initialize(writers_schema, readers_schema)
       end
     end
 
+    # Raised when a decoded array or map declares more elements than can be
+    # backed by the input, guarding against an out-of-memory attack.
+    class CollectionSizeError < AvroError; end
+
     # FIXME(jmhodges) move validate to this module?
 
     class BinaryDecoder
@@ -353,13 +357,14 @@ def read_enum(writers_schema, readers_schema, decoder)
       def read_array(writers_schema, readers_schema, decoder)
         read_items = []
         min_bytes = min_bytes_per_element(writers_schema.items)
+        total = 0
         block_count = decoder.read_long
         while block_count != 0
           if block_count < 0
             block_count = -block_count
             _block_size = decoder.read_long
           end
-          ensure_collection_available(decoder, block_count, min_bytes)
+          total = ensure_collection_available(decoder, total, block_count, min_bytes)
           block_count.times do
             read_items << read_data(writers_schema.items,
                                     readers_schema.items,
@@ -375,13 +380,14 @@ def read_map(writers_schema, readers_schema, decoder)
         read_items = {}
         # Map keys are strings (>= 1 byte length prefix) plus the value.
         min_bytes = 1 + min_bytes_per_element(writers_schema.values)
+        total = 0
         block_count = decoder.read_long
         while block_count != 0
           if block_count < 0
             block_count = -block_count
             _block_size = decoder.read_long
           end
-          ensure_collection_available(decoder, block_count, min_bytes)
+          total = ensure_collection_available(decoder, total, block_count, min_bytes)
           block_count.times do
             key = decoder.read_string
             read_items[key] = read_data(writers_schema.values,
@@ -526,11 +532,14 @@ def skip_union(writers_schema, decoder)
       end
 
       def skip_array(writers_schema, decoder)
-        skip_blocks(decoder) { skip_data(writers_schema.items, decoder) }
+        min_bytes = min_bytes_per_element(writers_schema.items)
+        skip_blocks(decoder, min_bytes) { skip_data(writers_schema.items, decoder) }
       end
 
       def skip_map(writers_schema, decoder)
-        skip_blocks(decoder) {
+        # Map keys are strings (>= 1 byte length prefix) plus the value.
+        min_bytes = 1 + min_bytes_per_element(writers_schema.values)
+        skip_blocks(decoder, min_bytes) {
           decoder.skip_string
           skip_data(writers_schema.values, decoder)
         }
@@ -566,30 +575,73 @@ def min_bytes_per_element(schema, visited = nil)
         end
       end
 
-      # Reject a collection (array or map) block whose declared element count
-      # could not be backed by the bytes actually remaining, before iterating.
-      # Skipped when the per-element minimum is zero or when the reader cannot
-      # report how many bytes remain.
-      def ensure_collection_available(decoder, count, min_bytes_per_element)
-        return if count <= 0 || min_bytes_per_element <= 0
-        # A decoder that implements the read protocol but not #bytes_remaining
-        # (e.g. a custom decoder) cannot report the remaining size; skip the
-        # check for it rather than raising NoMethodError.
-        return unless decoder.respond_to?(:bytes_remaining)
-        remaining = decoder.bytes_remaining
-        # Compare via integer division rather than multiplying, so a huge count
-        # does not create a large intermediate product.
-        if remaining && count > remaining / min_bytes_per_element
-          raise AvroError, "Collection claims #{count} elements with at least #{min_bytes_per_element} bytes each, but only #{remaining} bytes are available"
+      # Reject a collection (array or map) block that could drive an unbounded
+      # allocation, before iterating. A block whose declared element count could
+      # not be backed by the bytes actually remaining is rejected; a block of
+      # zero-byte elements (where the bytes-remaining check does not apply) is
+      # bounded by a cumulative item cap; and every collection is bounded by a
+      # structural cap. Returns the running total across blocks.
+      #
+      # Both limits default to the same values as the other Avro SDKs and can be
+      # overridden (to a single value capping both) via the
+      # AVRO_MAX_COLLECTION_ITEMS environment variable.
+      DEFAULT_MAX_COLLECTION_ITEMS = 10_000_000       # Zero-byte element cap.
+      DEFAULT_MAX_COLLECTION_STRUCTURAL = 2147483639  # Integer.MAX_VALUE - 8.
+
+      def collection_limits
+        env = Integer(ENV['AVRO_MAX_COLLECTION_ITEMS'], exception: false)
+        if env && env >= 0
+          [env, env]
+        else
+          [DEFAULT_MAX_COLLECTION_ITEMS, DEFAULT_MAX_COLLECTION_STRUCTURAL]
         end
       end
 
-      def skip_blocks(decoder, &blk)
+      def ensure_collection_available(decoder, total, count, min_bytes_per_element)
+        # A negative count is corrupt/malicious data (it can also arise from
+        # overflow when negating a negative block count); reject it explicitly.
+        raise CollectionSizeError, "Invalid negative collection block count: #{count}" if count < 0
+
+        total += count
+        max_items, max_structural = collection_limits
+
+        # A structural cap covers all collections, including decoders that cannot
+        # report their remaining bytes.
+        if total > max_structural
+          raise CollectionSizeError, "Collection size #{total} exceeds the maximum allowed size of #{max_structural}"
+        end
+
+        if min_bytes_per_element <= 0
+          # Zero-byte elements (e.g. null) consume no input, so the
+          # bytes-remaining check cannot bound them; cap by item count.
+          if total > max_items
+            raise CollectionSizeError, "Collection of zero-byte elements (#{total}) exceeds the maximum allowed size of #{max_items}"
+          end
+        elsif decoder.respond_to?(:bytes_remaining)
+          # A decoder that implements the read protocol but not #bytes_remaining
+          # (e.g. a custom decoder) cannot report the remaining size; skip the
+          # byte check for it rather than raising NoMethodError.
+          remaining = decoder.bytes_remaining
+          # Compare via integer division rather than multiplying, so a huge count
+          # does not create a large intermediate product.
+          if remaining && count > remaining / min_bytes_per_element
+            raise CollectionSizeError, "Collection claims #{count} elements with at least #{min_bytes_per_element} bytes each, but only #{remaining} bytes are available"
+          end
+        end
+
+        total
+      end
+
+      def skip_blocks(decoder, min_bytes = 0, &blk)
+        total = 0
         block_count = decoder.read_long
         while block_count != 0
           if block_count < 0
+            # A negative count carries a byte size, so the whole block can be
+            # skipped directly without iterating.
             decoder.skip(decoder.read_long)
           else
+            total = ensure_collection_available(decoder, total, block_count, min_bytes)
             block_count.times(&blk)
           end
           block_count = decoder.read_long
diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index 25314600006..23f7c6b29af 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -77,7 +77,7 @@ def decode(schema_json, encoded)
   def test_read_array_rejects_count_beyond_stream
     writer = StringIO.new
     Avro::IO::BinaryEncoder.new(writer).write_long(1_000_000)
-    assert_raise(Avro::AvroError) do
+    assert_raise(Avro::IO::CollectionSizeError) do
       decode('{"type":"array","items":"long"}', writer.string)
     end
   end
@@ -85,7 +85,7 @@ def test_read_array_rejects_count_beyond_stream
   def test_read_map_rejects_count_beyond_stream
     writer = StringIO.new
     Avro::IO::BinaryEncoder.new(writer).write_long(1_000_000)
-    assert_raise(Avro::AvroError) do
+    assert_raise(Avro::IO::CollectionSizeError) do
       decode('{"type":"map","values":"long"}', writer.string)
     end
   end
@@ -109,6 +109,44 @@ def test_read_array_within_stream_still_reads
     assert_equal([1, 2, 3], decode('{"type":"array","items":"long"}', writer.string))
   end
 
+  # A zero-byte element type (null) consumes no input, so the bytes-remaining
+  # check cannot bound its block count; a tiny payload declaring a huge count
+  # must instead be rejected by the item cap before building the array.
+  def test_read_array_of_null_rejects_huge_count
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(200_000_000) # ~4 byte payload
+    assert_raise(Avro::IO::CollectionSizeError) do
+      decode('{"type":"array","items":"null"}', writer.string)
+    end
+  end
+
+  def test_read_map_rejects_huge_count
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(200_000_000)
+    assert_raise(Avro::IO::CollectionSizeError) do
+      decode('{"type":"map","values":"long"}', writer.string)
+    end
+  end
+
+  # The skip path (a writer field absent from the reader schema) must be bounded
+  # too, so skipping a huge zero-byte block cannot loop endlessly.
+  def test_skip_array_of_null_rejects_huge_count
+    writers_schema = Avro::Schema.parse(<<-JSON)
+      {"type":"record","name":"Foo","fields":[
+        {"name":"arr","type":{"type":"array","items":"null"}},
+        {"name":"val","type":"int"}]}
+    JSON
+    readers_schema = Avro::Schema.parse(<<-JSON)
+      {"type":"record","name":"Foo","fields":[{"name":"val","type":"int"}]}
+    JSON
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(200_000_000) # arr block count
+    reader = Avro::IO::DatumReader.new(writers_schema, readers_schema)
+    assert_raise(Avro::IO::CollectionSizeError) do
+      reader.read(Avro::IO::BinaryDecoder.new(StringIO.new(writer.string)))
+    end
+  end
+
   def test_int
     check('"int"')
     check_default('"int"', "5", 5)

From 6b01deea0a6ececae394500d82e7169d282c5006 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 18:19:12 +0200
Subject: [PATCH 09/18] AVRO-4297: [ruby] Test INT64_MIN block count is bounded

Ruby integers do not overflow, so negating an INT64_MIN block count yields
2**63, which the existing zero-byte/structural cap already rejects. Add a
regression test decoding the 10-byte INT64_MIN varint as an array<null> block
count and asserting CollectionSizeError, matching the negation-overflow
coverage added to the C, C++ and C# SDKs.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/test/test_io.rb | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index 23f7c6b29af..926de4a13af 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -120,6 +120,17 @@ def test_read_array_of_null_rejects_huge_count
     end
   end
 
+  # INT64_MIN as a block count is the pathological negation case. Ruby integers
+  # do not overflow, so negating it yields 2**63, which the cap rejects. INT64_MIN
+  # zig-zag encodes as the 10-byte varint below, followed by a block byte-size (0)
+  # that the negative-block path reads.
+  def test_read_array_of_null_int64_min_block_count
+    payload = "\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\x01\x00".b
+    assert_raise(Avro::IO::CollectionSizeError) do
+      decode('{"type":"array","items":"null"}', payload)
+    end
+  end
+
   def test_read_map_rejects_huge_count
     writer = StringIO.new
     Avro::IO::BinaryEncoder.new(writer).write_long(200_000_000)

From f60163dbac3a7c57884b964fd8c8aebd7bb32c02 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 18:59:44 +0200
Subject: [PATCH 10/18] AVRO-4297: [ruby] Correct inaccurate negative-count
 comment

Ruby integers are arbitrary-precision, so negating a negative block count does
not overflow. Reword the comment in ensure_collection_available to reflect that
a negative count here only means corrupt/malicious data (the read/skip callers
already normalize a legitimate negative block count), avoiding confusion about a
Ruby overflow that cannot happen.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index eb632b24d4b..330c246f5f1 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -598,8 +598,9 @@ def collection_limits
       end
 
       def ensure_collection_available(decoder, total, count, min_bytes_per_element)
-        # A negative count is corrupt/malicious data (it can also arise from
-        # overflow when negating a negative block count); reject it explicitly.
+        # A negative count here is corrupt/malicious data (the read/skip callers
+        # already normalized a legitimate negative block count to its absolute
+        # value); reject it explicitly.
         raise CollectionSizeError, "Invalid negative collection block count: #{count}" if count < 0
 
         total += count

From 547dd67900d7a480520cdc773cc6b6f944fdd630 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 20:29:41 +0200
Subject: [PATCH 11/18] AVRO-4297: [ruby] Bound negative-block skip; reject
 negative block size

Addresses review feedback: skip_blocks applied ensure_collection_available only
for positive block counts, so the negative (byte-sized) block form could skip an
unbounded collection (e.g. array<null> with block size 0) and bypass the caps.
It now normalizes a negative count and bounds it like the positive path, and
rejects a negative block byte-size (which would seek the reader backwards) before
skipping. Adds a regression test for a huge negative-count skip block.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb  | 12 +++++++++---
 lang/ruby/test/test_io.rb | 21 +++++++++++++++++++++
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 330c246f5f1..8390374c2bb 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -638,9 +638,15 @@ def skip_blocks(decoder, min_bytes = 0, &blk)
         block_count = decoder.read_long
         while block_count != 0
           if block_count < 0
-            # A negative count carries a byte size, so the whole block can be
-            # skipped directly without iterating.
-            decoder.skip(decoder.read_long)
+            # A negative count declares abs(count) items preceded by a block
+            # byte-size. Bound the count too (so it can't bypass the caps), and
+            # reject a negative byte-size (which would seek the reader backwards)
+            # before skipping the whole block by its size.
+            block_count = -block_count
+            block_size = decoder.read_long
+            raise AvroError, "Invalid negative block size: #{block_size}" if block_size < 0
+            total = ensure_collection_available(decoder, total, block_count, min_bytes)
+            decoder.skip(block_size)
           else
             total = ensure_collection_available(decoder, total, block_count, min_bytes)
             block_count.times(&blk)
diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index 926de4a13af..2484df2d177 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -158,6 +158,27 @@ def test_skip_array_of_null_rejects_huge_count
     end
   end
 
+  def test_skip_array_of_null_negative_block_rejects_huge_count
+    # The negative (byte-sized) block form must also be bounded when skipping,
+    # so it cannot bypass the collection cap during projection.
+    writers_schema = Avro::Schema.parse(<<-JSON)
+      {"type":"record","name":"Foo","fields":[
+        {"name":"arr","type":{"type":"array","items":"null"}},
+        {"name":"val","type":"int"}]}
+    JSON
+    readers_schema = Avro::Schema.parse(<<-JSON)
+      {"type":"record","name":"Foo","fields":[{"name":"val","type":"int"}]}
+    JSON
+    writer = StringIO.new
+    encoder = Avro::IO::BinaryEncoder.new(writer)
+    encoder.write_long(-200_000_000) # negative block count
+    encoder.write_long(0)            # block byte-size
+    reader = Avro::IO::DatumReader.new(writers_schema, readers_schema)
+    assert_raise(Avro::IO::CollectionSizeError) do
+      reader.read(Avro::IO::BinaryDecoder.new(StringIO.new(writer.string)))
+    end
+  end
+
   def test_int
     check('"int"')
     check_default('"int"', "5", 5)

From ebf1f8be6c5aec151f8f0826411ab6139d84ac94 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 21:03:03 +0200
Subject: [PATCH 12/18] AVRO-4297: [ruby] Cap read_long varint length; validate
 skip block size

Addresses review feedback:
 - read_long bounded a 64-bit long to at most 10 bytes. Previously it consumed an
   arbitrarily long continuation chain, so a malicious input could force unbounded
   Integer growth (heavy CPU/memory) before any caller-side check ran.
 - skip_blocks now rejects a negative-block byte-size larger than the bytes
   remaining (when the reader can report them), so a truncated/malicious payload
   cannot seek the reader past EOF and later decode trailing zeros.

Adds regression tests for both.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb  | 15 ++++++++++++++-
 lang/ruby/test/test_io.rb | 28 ++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 8390374c2bb..4a23514fa87 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -75,8 +75,14 @@ def read_long
         b = byte!
         n = b & 0x7F
         shift = 7
+        # An Avro long is a 64-bit zig-zag varint, at most 10 bytes. Bound the
+        # continuation chain so a malicious input cannot force unbounded Integer
+        # growth (and heavy CPU/memory use) before any caller-side check runs.
+        count = 1
         while (b & 0x80) != 0
+          raise AvroError, "Varint is too long" if count >= 10
           b = byte!
+          count += 1
           n |= (b & 0x7F) << shift
           shift += 7
         end
@@ -641,10 +647,17 @@ def skip_blocks(decoder, min_bytes = 0, &blk)
             # A negative count declares abs(count) items preceded by a block
             # byte-size. Bound the count too (so it can't bypass the caps), and
             # reject a negative byte-size (which would seek the reader backwards)
-            # before skipping the whole block by its size.
+            # or one larger than the bytes remaining (a truncated input that would
+            # otherwise seek past EOF) before skipping the whole block by its size.
             block_count = -block_count
             block_size = decoder.read_long
             raise AvroError, "Invalid negative block size: #{block_size}" if block_size < 0
+            if decoder.respond_to?(:bytes_remaining)
+              remaining = decoder.bytes_remaining
+              if remaining && block_size > remaining
+                raise AvroError, "Cannot skip #{block_size} bytes, only #{remaining} remaining"
+              end
+            end
             total = ensure_collection_available(decoder, total, block_count, min_bytes)
             decoder.skip(block_size)
           else
diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index 2484df2d177..80cc025f8aa 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -179,6 +179,34 @@ def test_skip_array_of_null_negative_block_rejects_huge_count
     end
   end
 
+  def test_skip_negative_block_size_exceeding_remaining_is_rejected
+    # A negative block declares a byte-size; one larger than the bytes remaining
+    # would seek past EOF, so it must be rejected before skipping.
+    writers_schema = Avro::Schema.parse(<<-JSON)
+      {"type":"record","name":"Foo","fields":[
+        {"name":"arr","type":{"type":"array","items":"int"}},
+        {"name":"val","type":"int"}]}
+    JSON
+    readers_schema = Avro::Schema.parse(<<-JSON)
+      {"type":"record","name":"Foo","fields":[{"name":"val","type":"int"}]}
+    JSON
+    writer = StringIO.new
+    encoder = Avro::IO::BinaryEncoder.new(writer)
+    encoder.write_long(-1)        # negative block count: a byte-size follows
+    encoder.write_long(1_000_000) # block byte-size, but no data follows
+    reader = Avro::IO::DatumReader.new(writers_schema, readers_schema)
+    assert_raise(Avro::AvroError) do
+      reader.read(Avro::IO::BinaryDecoder.new(StringIO.new(writer.string)))
+    end
+  end
+
+  def test_read_long_rejects_overlong_varint
+    # An Avro long is at most 10 bytes; a longer continuation chain must be
+    # rejected rather than forcing unbounded Integer growth.
+    decoder = Avro::IO::BinaryDecoder.new(StringIO.new("\x80".b * 10))
+    assert_raise(Avro::AvroError) { decoder.read_long }
+  end
+
   def test_int
     check('"int"')
     check_default('"int"', "5", 5)

From e54959dec639de561d7a61073fdb9e2a27449184 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Sun, 12 Jul 2026 23:12:10 +0200
Subject: [PATCH 13/18] AVRO-4297: [ruby] Bounds-check union and enum indices

read_union and read_enum used the decoded index to index into the writer
schema's branches/symbols without validation. An out-of-range index returned
nil (raising a later NoMethodError), and because Ruby's negative indexing wraps,
a negative index silently selected the wrong branch/symbol.

Reject an index that is negative or >= the number of branches/symbols with a
clear AvroError before indexing. Adds tests for negative and too-large union and
enum indices.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb  |  8 ++++++++
 lang/ruby/test/test_io.rb | 37 +++++++++++++++++++++++++++++++++++++
 2 files changed, 45 insertions(+)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 4a23514fa87..077e95dfc98 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -349,6 +349,10 @@ def read_fixed(writers_schema, _readers_schema, decoder)
 
       def read_enum(writers_schema, readers_schema, decoder)
         index_of_symbol = decoder.read_int
+        if index_of_symbol < 0 || index_of_symbol >= writers_schema.symbols.size
+          raise AvroError, "Enum symbol index #{index_of_symbol} out of range " \
+                           "for #{writers_schema.symbols.size} symbols"
+        end
         read_symbol = writers_schema.symbols[index_of_symbol]
 
         if !readers_schema.symbols.include?(read_symbol) && readers_schema.default
@@ -408,6 +412,10 @@ def read_map(writers_schema, readers_schema, decoder)
 
       def read_union(writers_schema, readers_schema, decoder)
         index_of_schema = decoder.read_long
+        if index_of_schema < 0 || index_of_schema >= writers_schema.schemas.size
+          raise AvroError, "Union branch index #{index_of_schema} out of range " \
+                           "for #{writers_schema.schemas.size} branches"
+        end
         selected_writers_schema = writers_schema.schemas[index_of_schema]
 
         read_data(selected_writers_schema, readers_schema, decoder)
diff --git a/lang/ruby/test/test_io.rb b/lang/ruby/test/test_io.rb
index 80cc025f8aa..d8a0c8ed108 100644
--- a/lang/ruby/test/test_io.rb
+++ b/lang/ruby/test/test_io.rb
@@ -179,6 +179,43 @@ def test_skip_array_of_null_negative_block_rejects_huge_count
     end
   end
 
+  # A union branch index that is negative or beyond the number of branches is
+  # malformed and must be rejected (Ruby's negative indexing would otherwise
+  # silently select the wrong branch).
+  def test_read_union_rejects_out_of_range_index
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(5) # only 2 branches exist
+    assert_raise(Avro::AvroError) do
+      decode('["null","long"]', writer.string)
+    end
+  end
+
+  def test_read_union_rejects_negative_index
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_long(-1)
+    assert_raise(Avro::AvroError) do
+      decode('["null","long"]', writer.string)
+    end
+  end
+
+  # An enum symbol index that is negative or beyond the number of symbols is
+  # malformed and must be rejected.
+  def test_read_enum_rejects_out_of_range_index
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_int(9) # only 2 symbols exist
+    assert_raise(Avro::AvroError) do
+      decode('{"type":"enum","name":"E","symbols":["A","B"]}', writer.string)
+    end
+  end
+
+  def test_read_enum_rejects_negative_index
+    writer = StringIO.new
+    Avro::IO::BinaryEncoder.new(writer).write_int(-1)
+    assert_raise(Avro::AvroError) do
+      decode('{"type":"enum","name":"E","symbols":["A","B"]}', writer.string)
+    end
+  end
+
   def test_skip_negative_block_size_exceeding_remaining_is_rejected
     # A negative block declares a byte-size; one larger than the bytes remaining
     # would seek past EOF, so it must be rejected before skipping.

From 43f6233cb4def7fc97db1d88bf3fac2665df4dc4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 01:35:49 +0200
Subject: [PATCH 14/18] AVRO-4297: [ruby] Memoize collection_limits per reader
 instance

ensure_collection_available runs once per block, so parsing
ENV['AVRO_MAX_COLLECTION_ITEMS'] each time added avoidable overhead for
collections split into many blocks. Memoize the computed limits on the
DatumReader instance so the env var is read/parsed at most once per reader while
still letting a new reader instance pick up a changed value.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 077e95dfc98..edb3d39feb2 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -603,12 +603,19 @@ def min_bytes_per_element(schema, visited = nil)
       DEFAULT_MAX_COLLECTION_STRUCTURAL = 2147483639  # Integer.MAX_VALUE - 8.
 
       def collection_limits
-        env = Integer(ENV['AVRO_MAX_COLLECTION_ITEMS'], exception: false)
-        if env && env >= 0
-          [env, env]
-        else
-          [DEFAULT_MAX_COLLECTION_ITEMS, DEFAULT_MAX_COLLECTION_STRUCTURAL]
-        end
+        # Memoize per reader instance: ensure_collection_available runs once per
+        # block, so parsing the env var every time adds avoidable overhead for
+        # collections split into many blocks. Reading it once per reader still
+        # lets a new reader pick up a changed value.
+        @collection_limits ||=
+          begin
+            env = Integer(ENV['AVRO_MAX_COLLECTION_ITEMS'], exception: false)
+            if env && env >= 0
+              [env, env]
+            else
+              [DEFAULT_MAX_COLLECTION_ITEMS, DEFAULT_MAX_COLLECTION_STRUCTURAL]
+            end
+          end
       end
 
       def ensure_collection_available(decoder, total, count, min_bytes_per_element)

From db48317693e3583563eeb6c1aa9fb0a45da3aba6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 01:57:49 +0200
Subject: [PATCH 15/18] AVRO-4297: [ruby] Reject truncated reads; bounds-check
 skip_union index

Address review feedback:
- BinaryDecoder#read now raises AvroError when the underlying IO returns fewer
  bytes than requested (or nil), so a truncated/partial read for
  bytes/string/fixed is rejected instead of silently yielding corrupt data
  (which the "reject before allocating" hardening did not cover for small reads).
- skip_union validated no bounds on the decoded branch index; Ruby negative
  indexing would let a malformed index (e.g. -1) silently skip the wrong branch
  and desync the decoder. Apply the same [0, size) check as read_union.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index edb3d39feb2..9276469b0bf 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -136,7 +136,14 @@ def read(len)
             raise AvroError, "Cannot read #{len} bytes, only #{remaining} remaining"
           end
         end
-        @reader.read(len)
+        result = @reader.read(len)
+        # A truncated or partial read must not silently yield fewer bytes than
+        # requested (which would surface later as confusing corruption); reject it.
+        if len.positive? && (result.nil? || result.bytesize < len)
+          got = result.nil? ? 0 : result.bytesize
+          raise AvroError, "Truncated input: expected #{len} bytes, got #{got}"
+        end
+        result
       end
 
       # Number of bytes still available to read, or nil when the reader cannot
@@ -542,6 +549,10 @@ def skip_enum(_writers_schema, decoder)
 
       def skip_union(writers_schema, decoder)
         index = decoder.read_long
+        if index < 0 || index >= writers_schema.schemas.size
+          raise AvroError, "Union branch index #{index} out of range " \
+                           "for #{writers_schema.schemas.size} branches"
+        end
         skip_data(writers_schema.schemas[index], decoder)
       end
 

From b1863edd0a11f9ffd8500b40c04e1589b85fd62d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 02:17:53 +0200
Subject: [PATCH 16/18] AVRO-4297: [ruby] Bound skip_long to 10 bytes

read_long rejects overlong varints, but skip_long (used when projecting away a
long field) looped over continuation bytes with no cap, so a malicious payload
could force scanning unbounded input (CPU DoS) while skipping. Apply the same
10-byte limit and raise AvroError past it.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 9276469b0bf..4660fd52e0d 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -178,8 +178,13 @@ def skip_int
 
       def skip_long
         b = byte!
+        count = 1
         while (b & 0x80) != 0
+          # A 64-bit varint is at most 10 bytes; reject an overlong continuation
+          # chain so a skipped long can't force scanning unbounded input.
+          raise AvroError, "Varint is too long" if count >= 10
           b = byte!
+          count += 1
         end
       end
 

From 07166bbfbe5fb9b6c4c86f2bcd80d00ce816ff7f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 02:33:37 +0200
Subject: [PATCH 17/18] AVRO-4297: [ruby] Reject 10-byte varints outside the
 64-bit range

read_long capped varints at 10 bytes, but a 10-byte encoding can still set bits
beyond bit 63. The 10th byte contributes only bit 63, so reject it when any
higher payload bit (b & 0x7E) is set.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 4660fd52e0d..210c5f005f9 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -83,6 +83,9 @@ def read_long
           raise AvroError, "Varint is too long" if count >= 10
           b = byte!
           count += 1
+          # The 10th byte (count == 10) contributes only bit 63; any higher
+          # payload bit would push the value outside the 64-bit range.
+          raise AvroError, "Varint is too long" if count == 10 && (b & 0x7E) != 0
           n |= (b & 0x7F) << shift
           shift += 7
         end

From eb2ec9f64c80ae26240b593da9bf86de7b1c9f96 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= <iemejia@gmail.com>
Date: Mon, 13 Jul 2026 03:12:20 +0200
Subject: [PATCH 18/18] AVRO-4297: [ruby] skip_long 64-bit range check; reject
 negative read block sizes

Address review feedback:
- skip_long now also rejects a 10th varint byte with payload bits above bit 63,
  matching read_long, so a skipped long can't accept an out-of-64-bit-range
  encoding read_long would reject.
- read_array and read_map now reject a negative per-block byte-size in the
  negative-count form (as skip_blocks already does), instead of silently
  accepting malformed input.

Assisted-by: GitHub Copilot:claude-opus-4.8
---
 lang/ruby/lib/avro/io.rb | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/lang/ruby/lib/avro/io.rb b/lang/ruby/lib/avro/io.rb
index 210c5f005f9..a5b0a2961e2 100644
--- a/lang/ruby/lib/avro/io.rb
+++ b/lang/ruby/lib/avro/io.rb
@@ -188,6 +188,9 @@ def skip_long
           raise AvroError, "Varint is too long" if count >= 10
           b = byte!
           count += 1
+          # The 10th byte contributes only bit 63; reject out-of-64-bit-range
+          # encodings here too so skipping is consistent with read_long.
+          raise AvroError, "Varint is too long" if count == 10 && (b & 0x7E) != 0
         end
       end
 
@@ -387,7 +390,8 @@ def read_array(writers_schema, readers_schema, decoder)
         while block_count != 0
           if block_count < 0
             block_count = -block_count
-            _block_size = decoder.read_long
+            block_size = decoder.read_long
+            raise AvroError, "Invalid negative block size: #{block_size}" if block_size < 0
           end
           total = ensure_collection_available(decoder, total, block_count, min_bytes)
           block_count.times do
@@ -410,7 +414,8 @@ def read_map(writers_schema, readers_schema, decoder)
         while block_count != 0
           if block_count < 0
             block_count = -block_count
-            _block_size = decoder.read_long
+            block_size = decoder.read_long
+            raise AvroError, "Invalid negative block size: #{block_size}" if block_size < 0
           end
           total = ensure_collection_available(decoder, total, block_count, min_bytes)
           block_count.times do