Copilot commented on code in PR #3861:
URL: https://github.com/apache/avro/pull/3861#discussion_r3566596524
##########
lang/py/avro/io.py:
##########
@@ -795,24 +961,32 @@ def read_array(self, writers_schema:
avro.schema.ArraySchema, readers_schema: av
The actual count in this case
is the absolute value of the count written.
"""
- read_items = []
+ read_items: List[object] = []
+ min_bytes = _min_bytes_per_element(writers_schema.items)
+ zero_byte_limit, structural_limit = _collection_limits()
block_count = decoder.read_long()
while block_count != 0:
if block_count < 0:
block_count = -block_count
decoder.skip_long()
+ self._ensure_collection_available(decoder, len(read_items),
block_count, min_bytes, zero_byte_limit, structural_limit)
for i in range(block_count):
read_items.append(self.read_data(writers_schema.items,
readers_schema.items, decoder))
block_count = decoder.read_long()
return read_items
def skip_array(self, writers_schema: avro.schema.ArraySchema, decoder:
BinaryDecoder) -> None:
+ min_bytes = _min_bytes_per_element(writers_schema.items)
+ zero_byte_limit, structural_limit = _collection_limits()
+ items_skipped = 0
block_count = decoder.read_long()
while block_count != 0:
if block_count < 0:
block_size = decoder.read_long()
decoder.skip(block_size)
else:
+ self._ensure_collection_available(decoder, items_skipped,
block_count, min_bytes, zero_byte_limit, structural_limit)
Review Comment:
`skip_array()` applies `_ensure_collection_available()` only for positive
block counts. For negative block counts (the `-count` form with an explicit
`block_size`), the code currently bypasses both the bytes-remaining validation
and the cumulative item caps, so a malicious stream can evade the new
collection limit checks when the decoder is skipping data (e.g., for schema
resolution). Normalize negative counts and apply
`_ensure_collection_available()`/`items_skipped` consistently before skipping
the block bytes.
##########
lang/py/avro/io.py:
##########
@@ -832,25 +1006,34 @@ def read_map(self, writers_schema:
avro.schema.MapSchema, readers_schema: avro.s
The actual count in this case
is the absolute value of the count written.
"""
- read_items = {}
+ read_items: Dict[str, object] = {}
+ # Map keys are strings (>= 1 byte length prefix) plus the value.
+ min_bytes = 1 + _min_bytes_per_element(writers_schema.values)
+ zero_byte_limit, structural_limit = _collection_limits()
block_count = decoder.read_long()
while block_count != 0:
if block_count < 0:
block_count = -block_count
decoder.skip_long()
+ self._ensure_collection_available(decoder, len(read_items),
block_count, min_bytes, zero_byte_limit, structural_limit)
for i in range(block_count):
key = decoder.read_utf8()
read_items[key] = self.read_data(writers_schema.values,
readers_schema.values, decoder)
block_count = decoder.read_long()
return read_items
def skip_map(self, writers_schema: avro.schema.MapSchema, decoder:
BinaryDecoder) -> None:
+ min_bytes = 1 + _min_bytes_per_element(writers_schema.values)
+ zero_byte_limit, structural_limit = _collection_limits()
+ items_skipped = 0
block_count = decoder.read_long()
while block_count != 0:
if block_count < 0:
block_size = decoder.read_long()
decoder.skip(block_size)
else:
+ self._ensure_collection_available(decoder, items_skipped,
block_count, min_bytes, zero_byte_limit, structural_limit)
Review Comment:
`skip_map()` has the same gap as `skip_array()`: negative block counts
bypass `_ensure_collection_available()` and the cumulative `items_skipped`
accounting. This makes the new size-limit enforcement inconsistent between read
vs skip paths and allows the cap to be bypassed when skipping map blocks
encoded with negative counts.
##########
lang/py/avro/test/test_io.py:
##########
@@ -715,6 +843,165 @@ def test_type_exception_record(self) -> None:
write_datum(datum_to_write, writers_schema)
+class TestDatumReaderCollectionSizeLimit(unittest.TestCase):
+ """Elements whose schema encodes to zero bytes (``null``, a zero-length
+ ``fixed``, or a record with only zero-byte fields) consume no input, so the
+ bytes-remaining check cannot bound them. A huge declared block count of
such
+ elements is capped so a tiny payload cannot exhaust memory. The limit is
+ configurable via the ``AVRO_MAX_COLLECTION_ITEMS`` environment variable."""
+
+ @staticmethod
+ def _decode(schema_json: str, encoded: bytes) -> object:
+ schema = avro.schema.parse(schema_json)
+ reader = avro.io.DatumReader(schema)
+ with io.BytesIO(encoded) as bio:
+ return reader.read(avro.io.BinaryDecoder(bio))
+
+ @staticmethod
+ def _skip(schema_json: str, encoded: bytes) -> None:
+ schema = avro.schema.parse(schema_json)
+ reader = avro.io.DatumReader(schema)
+ with io.BytesIO(encoded) as bio:
+ reader.skip_data(schema, avro.io.BinaryDecoder(bio))
+
+ @staticmethod
+ def _array_block(count: int, *, negative: bool = False) -> bytes:
+ """One array/map block of ``count`` zero-byte elements + end marker."""
+ buf = io.BytesIO()
+ enc = avro.io.BinaryEncoder(buf)
+ if negative:
+ enc.write_long(-count) # negative count is followed by a block
byte-size
+ enc.write_long(0)
+ else:
+ enc.write_long(count)
+ enc.write_long(0) # end-of-collection marker
+ return buf.getvalue()
+
+ def test_array_of_null_exceeds_default_limit(self) -> None:
+ # The reported exploit: a ~6 byte payload declaring 200,000,000 nulls.
+ self.assertRaises(
+ avro.errors.AvroCollectionSizeException,
+ self._decode,
+ '{"type": "array", "items": "null"}',
+ self._array_block(200_000_000),
+ )
+
+ def test_array_of_null_within_configured_limit_still_reads(self) -> None:
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ result = self._decode('{"type": "array", "items": "null"}',
self._array_block(1000))
+ self.assertEqual(result, [None] * 1000)
+
+ def test_array_of_null_exceeds_configured_limit(self) -> None:
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ self.assertRaises(
+ avro.errors.AvroCollectionSizeException,
+ self._decode,
+ '{"type": "array", "items": "null"}',
+ self._array_block(1001),
+ )
+
+ def test_array_of_null_cumulative_across_blocks(self) -> None:
+ # Two blocks of 600 nulls each (1200 > 1000) must be rejected on the
second.
+ buf = io.BytesIO()
+ enc = avro.io.BinaryEncoder(buf)
+ enc.write_long(600)
+ enc.write_long(600)
+ enc.write_long(0)
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ self.assertRaises(
+ avro.errors.AvroCollectionSizeException,
+ self._decode,
+ '{"type": "array", "items": "null"}',
+ buf.getvalue(),
+ )
+
+ def test_array_of_null_negative_block_count(self) -> None:
+ # A negative count encodes abs(count) elements preceded by a block
size;
+ # after normalization it must still be bounded.
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ self.assertRaises(
+ avro.errors.AvroCollectionSizeException,
+ self._decode,
+ '{"type": "array", "items": "null"}',
+ self._array_block(200_000, negative=True),
+ )
+
+ def test_array_of_zero_length_fixed_exceeds_limit(self) -> None:
+ schema_json = '{"type": "array", "items": {"type": "fixed", "name":
"empty", "size": 0}}'
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ self.assertRaises(avro.errors.AvroCollectionSizeException,
self._decode, schema_json, self._array_block(2000))
+
+ def test_array_of_all_null_record_exceeds_limit(self) -> None:
+ schema_json = '{"type": "array", "items": {"type": "record", "name":
"R", "fields": [{"name": "n", "type": "null"}]}}'
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ self.assertRaises(avro.errors.AvroCollectionSizeException,
self._decode, schema_json, self._array_block(2000))
+
+ def test_map_of_null_rejected_by_available_bytes(self) -> None:
+ # Map entries always carry a >= 1 byte key, so a huge map<null> is
bounded
+ # by the bytes-remaining check rather than the zero-byte cap.
+ self.assertRaises(
+ avro.errors.InvalidAvroBinaryEncoding,
+ self._decode,
+ '{"type": "map", "values": "null"}',
+ self._array_block(200_000_000),
+ )
+
+ def test_skip_array_of_null_respects_limit(self) -> None:
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1000"}):
+ self.assertRaises(
+ avro.errors.AvroCollectionSizeException,
+ self._skip,
+ '{"type": "array", "items": "null"}',
+ self._array_block(2000),
+ )
+
+ def test_invalid_env_override_falls_back_to_default(self) -> None:
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "not-a-number"}):
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore")
+ self.assertEqual(avro.io._max_collection_items(),
avro.io.DEFAULT_MAX_COLLECTION_ITEMS)
+
+ def test_negative_env_override_falls_back_to_default(self) -> None:
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "-5"}):
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore")
+ self.assertEqual(avro.io._max_collection_items(),
avro.io.DEFAULT_MAX_COLLECTION_ITEMS)
+
+ def test_env_override_is_honored(self) -> None:
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "7"}):
+ self.assertEqual(avro.io._max_collection_items(), 7)
+ result = self._decode('{"type": "array", "items": "null"}',
self._array_block(7))
+ self.assertEqual(result, [None] * 7)
+
+ def test_collection_limits_env_caps_both(self) -> None:
+ # When set, AVRO_MAX_COLLECTION_ITEMS caps both the zero-byte and the
+ # structural limit; unset, they differ (tighter zero-byte default).
+ with unittest.mock.patch.dict(os.environ,
{"AVRO_MAX_COLLECTION_ITEMS": "1234"}):
+ self.assertEqual(avro.io._collection_limits(), (1234, 1234))
+ os.environ.pop("AVRO_MAX_COLLECTION_ITEMS", None)
+ self.assertEqual(
+ avro.io._collection_limits(),
+ (avro.io.DEFAULT_MAX_COLLECTION_ITEMS,
avro.io.DEFAULT_MAX_COLLECTION_STRUCTURAL),
+ )
Review Comment:
`test_collection_limits_env_caps_both()` mutates the real process
environment via `os.environ.pop(...)` outside of a patch context. If a
developer (or CI) runs tests with `AVRO_MAX_COLLECTION_ITEMS` already set, this
test will delete it for the remainder of the test run, potentially causing
unrelated tests to behave differently. Save/restore the original value (or keep
the unset assertion inside a temporary env override) so the test is hermetic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
