From 3dbcb4329b682c36db4fc42d29e3c21858fc15a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 6 Aug 2026 11:05:31 +0200 Subject: [PATCH] AVRO-4296: [python] Bound allocation when decoding length-prefixed values and collections When decoding an array or map, DatumReader.read_array/read_map used the block count read from the (potentially malformed or truncated) input directly as a loop counter, and the length-prefixed byte/string readers allocated from an unchecked length. A small payload could therefore declare a very large count or length and drive an unbounded allocation before any element bytes were present. This bounds those allocations, mirroring the Java SDK's two-limit approach: - Length-prefixed values and collection blocks are validated against the number of bytes actually remaining in the input (when the reader is seekable) before allocating, so a truncated payload fails fast instead of over-allocating. - Element types whose minimum encoded size is zero (null, a zero-length fixed, or a record whose fields are all zero-byte) cannot be bounded by the bytes remaining, so their cumulative count is capped by a separate configurable limit (AVRO_MAX_COLLECTION_ITEMS). A structural cap of Integer.MAX_VALUE - 8 applies to all collections as defense in depth. It also hardens related decode paths surfaced by this work: overlong/overflowing varints are rejected, negative skips and out-of-range union/enum indices are rejected, sized-block skips validate their byte size, and bytes_remaining() restores the reader position and degrades gracefully on non-seekable readers. Reading malformed or truncated input now fails fast with a clear AvroCollectionSizeException or InvalidAvroBinaryEncoding; valid data reads unchanged. --- lang/py/avro/errors.py | 9 + lang/py/avro/io.py | 328 ++++++++++++++++++++++- lang/py/avro/test/test_io.py | 485 +++++++++++++++++++++++++++++++++++ 3 files changed, 814 insertions(+), 8 deletions(-) diff --git a/lang/py/avro/errors.py b/lang/py/avro/errors.py index b961a04ae43..3939a364fa5 100644 --- a/lang/py/avro/errors.py +++ b/lang/py/avro/errors.py @@ -36,6 +36,15 @@ class InvalidAvroBinaryEncoding(AvroException): """For invalid numbers of bytes read.""" +class AvroCollectionSizeException(AvroException): + """Raised when a decoded array or map would exceed a collection size limit. + + This covers both the zero-byte-element cap (the cumulative number of + zero-byte elements such as ``null``) and the structural cap on the total + number of elements in any collection. + """ + + class SchemaParseException(AvroException): """Raised when a schema failed to parse.""" diff --git a/lang/py/avro/io.py b/lang/py/avro/io.py index d0d9a738eb8..a67230f0734 100644 --- a/lang/py/avro/io.py +++ b/lang/py/avro/io.py @@ -87,9 +87,10 @@ import collections import datetime import decimal +import os import struct import warnings -from typing import IO, Generator, Iterable, List, Mapping, Optional, Sequence, Union +from typing import IO, Dict, Generator, Iterable, List, Mapping, Optional, Sequence, Set, Tuple, Union import avro.constants import avro.errors @@ -198,6 +199,13 @@ class BinaryDecoder: _reader: IO[bytes] + # Reads with a declared length above this many bytes are validated against + # the number of bytes actually remaining (when the reader is seekable) + # 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, since they cannot cause a meaningful over-allocation. + _MAX_UNCHECKED_READ = 1024 * 1024 + def __init__(self, reader: IO[bytes]) -> None: """ reader is a Python object on which we can call read, seek, and tell. @@ -214,11 +222,46 @@ def read(self, n: int) -> bytes: """ if n < 0: raise avro.errors.InvalidAvroBinaryEncoding(f"Requested {n} bytes to read, expected positive integer.") + if n > self._MAX_UNCHECKED_READ: + remaining = self.bytes_remaining() + if remaining is not None and n > remaining: + raise avro.errors.InvalidAvroBinaryEncoding(f"Requested {n} bytes to read, but only {remaining} remain.") read_bytes = self.reader.read(n) if len(read_bytes) != n: raise avro.errors.InvalidAvroBinaryEncoding(f"Read {len(read_bytes)} bytes, expected {n} bytes") return read_bytes + def bytes_remaining(self) -> Optional[int]: + """ + Return the number of bytes still available to read, or ``None`` when + that count is not known (a non-seekable reader, or one whose position + cannot be obtained). Used to reject a declared length or collection + block count that exceeds the data actually available before allocating + for it. + """ + reader = self.reader + try: + pos = reader.tell() + except (OSError, ValueError, AttributeError, TypeError): + # Not seekable, or the position could not be determined. + return None + try: + reader.seek(0, os.SEEK_END) + end = reader.tell() + # Clamp to 0: a stream positioned past its end (seekable file-like + # objects allow seeking beyond EOF) would otherwise report a negative + # "remaining", violating the non-negative contract. + return max(0, end - pos) + except (OSError, ValueError, AttributeError, TypeError): + return None + finally: + # Always restore the original position, even if seeking to the end + # or reading it failed, so the reader is never left at EOF. + try: + reader.seek(pos) + except (OSError, ValueError, AttributeError, TypeError): + pass + def read_null(self) -> None: """ null is written as zero bytes @@ -246,7 +289,17 @@ def read_long(self) -> int: n = b & 0x7F shift = 7 while (b & 0x80) != 0: + # A 64-bit value needs at most 10 bytes (shifts 0..63); reject an + # overlong varint rather than accepting a malformed, arbitrarily + # large value. + if shift >= 70: + raise avro.errors.InvalidAvroBinaryEncoding("Varint is too long") b = ord(self.read(1)) + # The 10th byte (shift == 63) contributes only bit 63; any higher + # payload bit would push the value outside the 64-bit range, so a + # valid zig-zag long must have them clear. + if shift == 63 and (b & 0x7E) != 0: + raise avro.errors.InvalidAvroBinaryEncoding("Varint is too long") n |= (b & 0x7F) << shift shift += 7 datum = (n >> 1) ^ -(n & 1) @@ -383,8 +436,21 @@ def skip_int(self) -> None: def skip_long(self) -> None: b = ord(self.read(1)) + shift = 7 while (b & 0x80) != 0: + # A 64-bit varint is at most 10 bytes; reject an overlong chain so a + # skipped long can't force scanning unbounded input (read_long caps + # the same way). + if shift >= 70: + raise avro.errors.InvalidAvroBinaryEncoding("Varint is too long") b = ord(self.read(1)) + # The 10th byte (shift == 63) contributes only bit 63; any higher + # payload bit would push the value outside the 64-bit range. Reject + # it here too so skip_long enforces the same validity as read_long + # (e.g. a malformed negative-block byte-size in read_array/read_map). + if shift == 63 and (b & 0x7E) != 0: + raise avro.errors.InvalidAvroBinaryEncoding("Varint is too long") + shift += 7 def skip_float(self) -> None: self.skip(4) @@ -393,12 +459,26 @@ def skip_double(self) -> None: self.skip(8) def skip_bytes(self) -> None: - self.skip(self.read_long()) + # The length prefix is attacker-controlled: an oversized value would seek + # past EOF (a negative one is rejected by ``skip`` below, which is the + # single place that guards backward seeks). Validate an oversized length + # against the bytes remaining the same way ``read`` does before skipping. + n = self.read_long() + if n > self._MAX_UNCHECKED_READ: + remaining = self.bytes_remaining() + if remaining is not None and n > remaining: + raise avro.errors.InvalidAvroBinaryEncoding(f"Requested {n} bytes to skip, but only {remaining} remain.") + self.skip(n) def skip_utf8(self) -> None: self.skip_bytes() def skip(self, n: int) -> None: + # Guard against a negative skip (a backward seek), which would corrupt + # the decoder position. Callers pass schema-derived sizes (fixed/float/ + # double) or already-validated lengths, so this is a defensive backstop. + if n < 0: + raise avro.errors.InvalidAvroBinaryEncoding(f"Cannot skip a negative number of bytes: {n}") self.reader.seek(self.reader.tell() + n) @@ -599,6 +679,118 @@ def write_timestamp_micros_long(self, datum: datetime.datetime) -> None: # # DatumReader/Writer # + +# Environment variable overriding the collection element limits. When set to a +# non-negative integer it caps both the number of zero-byte-encoded collection +# elements (e.g. an array of nulls) and the structural cap on the total number of +# elements in any collection, allocated from a single decode. +MAX_COLLECTION_ITEMS_ENV = "AVRO_MAX_COLLECTION_ITEMS" + +# Default maximum number of zero-byte-encoded collection elements to allocate. +# 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 their count; without a cap a tiny payload can declare a huge +# block count and exhaust memory. A legitimate collection of zero-byte elements is +# small, so this default is generous while still rejecting pathological input. It +# can be raised (or lowered) with the ``AVRO_MAX_COLLECTION_ITEMS`` environment +# variable. +DEFAULT_MAX_COLLECTION_ITEMS = 10_000_000 + +# Default structural cap on the number of elements in any array or map (a +# collection larger than this is treated as malformed regardless of element +# type). Matches the historical ``Integer.MAX_VALUE - 8`` limit. Elements with a +# positive on-wire size are also bounded by the bytes remaining; this cap is an +# additional overflow/defense-in-depth guard. ``AVRO_MAX_COLLECTION_ITEMS``, when +# set, overrides both this and ``DEFAULT_MAX_COLLECTION_ITEMS``. +DEFAULT_MAX_COLLECTION_STRUCTURAL = (1 << 31) - 1 - 8 # Integer.MAX_VALUE - 8 + +# Block counts at or below this are not checked against the bytes remaining. A +# collection element with a positive on-wire size must be backed by real bytes, +# so a small block cannot over-allocate meaningfully; skipping the check avoids a +# per-block ``bytes_remaining()`` seek (tell + seek-to-end + seek-back), which is +# not free on a real file object. Mirrors ``BinaryDecoder._MAX_UNCHECKED_READ``. +# The structural cap below is always enforced (no seek), and zero-byte elements +# keep their own cumulative cap. +_MAX_UNCHECKED_COLLECTION = 1024 + + +def _collection_limits() -> Tuple[int, int]: + """Return ``(zero_byte_limit, structural_limit)``. + + ``AVRO_MAX_COLLECTION_ITEMS``, when set to a non-negative integer, pins *both* + limits to that single value. The coupling runs in both directions: raising it + to lift the zero-byte-element limit also lowers the structural cap from + ``DEFAULT_MAX_COLLECTION_STRUCTURAL`` (~2.1 billion) to that same value, and + lowering it tightens both. Set it above ``DEFAULT_MAX_COLLECTION_STRUCTURAL`` + if you need to raise the zero-byte limit without reducing the structural cap. + When unset, zero-byte elements use the tighter ``DEFAULT_MAX_COLLECTION_ITEMS`` + and all collections use ``DEFAULT_MAX_COLLECTION_STRUCTURAL``. + """ + value = os.environ.get(MAX_COLLECTION_ITEMS_ENV) + if value is None: + return DEFAULT_MAX_COLLECTION_ITEMS, DEFAULT_MAX_COLLECTION_STRUCTURAL + try: + parsed = int(value) + except ValueError: + warnings.warn(avro.errors.AvroWarning(f"Ignoring invalid {MAX_COLLECTION_ITEMS_ENV} value: {value!r}")) + return DEFAULT_MAX_COLLECTION_ITEMS, DEFAULT_MAX_COLLECTION_STRUCTURAL + if parsed < 0: + warnings.warn(avro.errors.AvroWarning(f"Ignoring negative {MAX_COLLECTION_ITEMS_ENV} value: {value!r}")) + return DEFAULT_MAX_COLLECTION_ITEMS, DEFAULT_MAX_COLLECTION_STRUCTURAL + return parsed, parsed + + +def _max_collection_items() -> int: + """Return the configured zero-byte-element collection limit.""" + return _collection_limits()[0] + + +def _min_bytes_per_element(schema: avro.schema.Schema, visited: Optional[Set[int]] = None) -> int: + """ + Return the minimum number of bytes a single value of ``schema`` can occupy + on the wire. Used to reject an array/map block count that could not possibly + be backed by the bytes remaining. A type that can encode to zero bytes + (``null``) returns 0, which disables the collection check for it (avoiding a + false positive on, e.g., an array of nulls). + """ + if visited is None: + visited = set() + schema_type = schema.type + if schema_type == "null": + return 0 + if schema_type == "float": + return 4 + if schema_type == "double": + return 8 + if schema_type == "fixed": + return getattr(schema, "size", 1) + if schema_type in ("record", "error"): + # Guard against self-referencing records (recursion would not terminate). + # A recursive reference is not a zero-byte value: any finite recursive + # value must terminate through a union (>= 1 byte branch index) or an + # empty array/map (>= 1 byte block count), so 1 is a safe conservative + # lower bound. Returning 0 here would wrongly treat recursive records as + # zero-byte elements and weaken the bytes-remaining precheck. + if id(schema) in visited: + return 1 + visited.add(id(schema)) + total = 0 + for field in getattr(schema, "fields", []): + total += _min_bytes_per_element(field.type, visited) + visited.discard(id(schema)) + return total + if schema_type == "union": + # A union encodes a >= 1 byte branch index plus the selected branch's + # payload, so its minimum is 1 + the smallest branch minimum (which is + # 1 when any branch is null / zero-byte). + branches = getattr(schema, "schemas", []) + if not branches: + return 1 + return 1 + min(_min_bytes_per_element(branch, visited) for branch in branches) + # boolean, int, long, bytes, string, enum, array, map: all >= 1 byte. + return 1 + + class DatumReader: """Deserialize Avro-encoded data into a Python data structure.""" @@ -765,7 +957,14 @@ def read_enum(self, writers_schema: avro.schema.EnumSchema, readers_schema: avro """ # read data index_of_symbol = decoder.read_int() + if index_of_symbol < 0: + # A negative index is malformed data and never resolves to a default. + raise avro.errors.SchemaResolutionException( + f"Can't access enum index {index_of_symbol} for enum with {len(writers_schema.symbols)} symbols", writers_schema, readers_schema + ) if index_of_symbol >= len(writers_schema.symbols): + # A symbol beyond the writer's schema may be a future symbol; fall + # back to the schema's default when one is defined. default = writers_schema.default if default is not None: return default @@ -783,6 +982,54 @@ def read_enum(self, writers_schema: avro.schema.EnumSchema, readers_schema: avro def skip_enum(self, writers_schema: avro.schema.EnumSchema, decoder: BinaryDecoder) -> None: return decoder.skip_int() + @staticmethod + def _ensure_collection_available( + decoder: BinaryDecoder, + existing: int, + count: int, + min_bytes_per_element: int, + zero_byte_limit: int, + structural_limit: int, + ) -> None: + """ + Reject a collection (array or map) block that could not be backed by the + input, before iterating. + + For elements with a positive minimum on-wire size, the declared count is + checked against the bytes actually remaining and against a structural + limit (an overflow/defense-in-depth cap). For zero-byte elements (e.g. an + array of nulls), which consume no input and so cannot be bounded by the + bytes remaining, the cumulative count is checked against the (tighter) + zero-byte limit. + """ + if count <= 0: + return + if min_bytes_per_element > 0: + # Only pay for the bytes_remaining() seek on a large block: a small + # block of positive-size elements must be backed by real bytes on the + # wire and so cannot over-allocate meaningfully (see + # _MAX_UNCHECKED_COLLECTION). The structural cap below is always + # enforced without a seek. + if count > _MAX_UNCHECKED_COLLECTION: + remaining = decoder.bytes_remaining() + # Compare via integer division rather than multiplying, so an + # attacker-controlled (unbounded) count does not create a huge + # intermediate product. + if remaining is not None and count > remaining // min_bytes_per_element: + raise avro.errors.InvalidAvroBinaryEncoding( + f"Collection claims {count} elements with at least {min_bytes_per_element} bytes each, but only {remaining} bytes are available." + ) + if existing + count > structural_limit: + raise avro.errors.AvroCollectionSizeException( + f"Cannot read a collection of more than {structural_limit} elements " + f"(declared {existing + count}); raise the {MAX_COLLECTION_ITEMS_ENV} limit if this is legitimate." + ) + elif existing + count > zero_byte_limit: + raise avro.errors.AvroCollectionSizeException( + f"Cannot read a collection of more than {zero_byte_limit} zero-byte elements " + f"(declared {existing + count}); raise the {MAX_COLLECTION_ITEMS_ENV} limit if this is legitimate." + ) + def read_array(self, writers_schema: avro.schema.ArraySchema, readers_schema: avro.schema.ArraySchema, decoder: BinaryDecoder) -> List[object]: """ Arrays are encoded as a series of blocks. @@ -798,23 +1045,69 @@ 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 + @staticmethod + def _skip_block_bytes(decoder: BinaryDecoder, block_size: int, block_count: int, min_bytes: int) -> None: + """Skip a sized-block's byte count, rejecting malformed sizes. + + The block_size is attacker-controlled: a negative value would seek + backwards and an oversized value past EOF, either corrupting the decoder + position. Also require that block_size can plausibly hold block_count + elements at their minimum on-wire size, so a too-small size cannot + misalign the decoder. A zero-byte element type (``null``, a zero-length + ``fixed``, or a record of only zero-byte fields) encodes to exactly 0 + bytes, so its block payload must be empty; a positive size would skip + into the following fields. Reject before skipping. + """ + if block_size < 0: + raise avro.errors.InvalidAvroBinaryEncoding(f"Invalid negative block size: {block_size}") + remaining = decoder.bytes_remaining() + if remaining is not None and block_size > remaining: + raise avro.errors.InvalidAvroBinaryEncoding(f"Block size {block_size} exceeds the {remaining} bytes remaining") + if min_bytes > 0: + if block_count > block_size // min_bytes: + raise avro.errors.InvalidAvroBinaryEncoding( + f"Block size {block_size} is too small for {block_count} elements of >= {min_bytes} bytes" + ) + elif block_size != 0: + raise avro.errors.InvalidAvroBinaryEncoding(f"Block size {block_size} must be 0 for a zero-byte element type") + try: + decoder.skip(block_size) + except (OSError, ValueError, OverflowError, AttributeError, TypeError) as e: + # The underlying reader can reject the seek (oversized offset, invalid + # seek, or other IO error), or lack seek()/tell() entirely; surface it + # as an Avro decoding error rather than leaking a non-Avro exception. + raise avro.errors.InvalidAvroBinaryEncoding(f"Cannot skip block of {block_size} bytes: {e}") from e + 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: + block_size = None if block_count < 0: + block_count = -block_count block_size = decoder.read_long() - decoder.skip(block_size) + # Bound the (normalized) count on both the sized and unsized paths so + # a negative block count cannot bypass the collection limits. + self._ensure_collection_available(decoder, items_skipped, block_count, min_bytes, zero_byte_limit, structural_limit) + items_skipped += block_count + if block_size is not None: + self._skip_block_bytes(decoder, block_size, block_count, min_bytes) else: for i in range(block_count): self.skip_data(writers_schema.items, decoder) @@ -835,12 +1128,21 @@ 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() + # Track decoded pairs separately: len(read_items) counts unique keys, so + # duplicate keys (later entries overwrite earlier ones) would undercount + # and let a multi-block map exceed the cumulative caps. + items_read = 0 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, items_read, block_count, min_bytes, zero_byte_limit, structural_limit) + items_read += block_count for i in range(block_count): key = decoder.read_utf8() read_items[key] = self.read_data(writers_schema.values, readers_schema.values, decoder) @@ -848,11 +1150,21 @@ def read_map(self, writers_schema: avro.schema.MapSchema, readers_schema: avro.s 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: + block_size = None if block_count < 0: + block_count = -block_count block_size = decoder.read_long() - decoder.skip(block_size) + # Bound the (normalized) count on both the sized and unsized paths so + # a negative block count cannot bypass the collection limits. + self._ensure_collection_available(decoder, items_skipped, block_count, min_bytes, zero_byte_limit, structural_limit) + items_skipped += block_count + if block_size is not None: + self._skip_block_bytes(decoder, block_size, block_count, min_bytes) else: for i in range(block_count): decoder.skip_utf8() @@ -867,7 +1179,7 @@ def read_union(self, writers_schema: avro.schema.UnionSchema, readers_schema: av """ # schema resolution index_of_schema = int(decoder.read_long()) - if index_of_schema >= len(writers_schema.schemas): + if index_of_schema < 0 or index_of_schema >= len(writers_schema.schemas): raise avro.errors.SchemaResolutionException( f"Can't access branch index {index_of_schema} for union with {len(writers_schema.schemas)} branches", writers_schema, readers_schema ) @@ -878,7 +1190,7 @@ def read_union(self, writers_schema: avro.schema.UnionSchema, readers_schema: av def skip_union(self, writers_schema: avro.schema.UnionSchema, decoder: BinaryDecoder) -> None: index_of_schema = int(decoder.read_long()) - if index_of_schema >= len(writers_schema.schemas): + if index_of_schema < 0 or index_of_schema >= len(writers_schema.schemas): raise avro.errors.SchemaResolutionException( f"Can't access branch index {index_of_schema} for union with {len(writers_schema.schemas)} branches", writers_schema ) diff --git a/lang/py/avro/test/test_io.py b/lang/py/avro/test/test_io.py index 41d7f827cfa..9ee39b1937b 100644 --- a/lang/py/avro/test/test_io.py +++ b/lang/py/avro/test/test_io.py @@ -23,7 +23,9 @@ import io import itertools import json +import os import unittest +import unittest.mock import uuid import warnings from typing import BinaryIO, Collection, Dict, List, Optional, Tuple, Union, cast @@ -550,6 +552,209 @@ def test_deserialization_fails(self) -> None: ) +class TestBinaryDecoderAvailableBytes(unittest.TestCase): + """A bytes/string value declares a length prefix; a malicious or truncated + input can declare far more bytes than actually exist. On a seekable reader + that is rejected before allocating for it.""" + + @staticmethod + def _encode_length_prefix(length: int) -> bytes: + buf = io.BytesIO() + avro.io.BinaryEncoder(buf).write_long(length) + return buf.getvalue() + + def test_read_rejects_length_beyond_stream(self) -> None: + # Declares 100 MiB but provides no data. + prefix = self._encode_length_prefix(100 * 1024 * 1024) + with io.BytesIO(prefix) as bio: + decoder = avro.io.BinaryDecoder(bio) + self.assertRaises(avro.errors.InvalidAvroBinaryEncoding, decoder.read_bytes) + + def test_read_within_stream_still_reads(self) -> None: + # A well-formed large value whose data is actually present still reads. + payload = b"x" * (2 * 1024 * 1024) + buf = io.BytesIO() + avro.io.BinaryEncoder(buf).write_bytes(payload) + with io.BytesIO(buf.getvalue()) as bio: + decoder = avro.io.BinaryDecoder(bio) + self.assertEqual(decoder.read_bytes(), payload) + + def test_bytes_remaining_restores_position(self) -> None: + # bytes_remaining() must leave the reader position unchanged. + with io.BytesIO(b"abcdefghij") as bio: + bio.seek(3) + decoder = avro.io.BinaryDecoder(bio) + self.assertEqual(decoder.bytes_remaining(), 7) + self.assertEqual(bio.tell(), 3) + + def test_bytes_remaining_restores_position_on_error(self) -> None: + # If reading the end offset fails after seeking, the original position + # must still be restored (via the finally block). + class FailingEndStream(io.BytesIO): + def seek(self, offset: int, whence: int = os.SEEK_SET) -> int: + # Fail only when seeking to the end. + if whence == os.SEEK_END: + raise OSError("cannot seek to end") + return super().seek(offset, whence) + + stream = FailingEndStream(b"abcdefghij") + stream.seek(4) + decoder = avro.io.BinaryDecoder(stream) + self.assertIsNone(decoder.bytes_remaining()) + self.assertEqual(stream.tell(), 4) + + def test_read_non_seekable_falls_back(self) -> None: + # A non-seekable reader skips the pre-check, but the post-read length + # check still rejects a truncated oversized value. + prefix = self._encode_length_prefix(100 * 1024 * 1024) + + class NonSeekable: + def __init__(self, data: bytes) -> None: + self._bio = io.BytesIO(data) + + def read(self, n: int = -1) -> bytes: + return self._bio.read(n) + + def seekable(self) -> bool: + return False + + decoder = avro.io.BinaryDecoder(NonSeekable(prefix)) # type: ignore[arg-type] + self.assertRaises(avro.errors.InvalidAvroBinaryEncoding, decoder.read_bytes) + + def test_skip_bytes_rejects_negative_length(self) -> None: + # A negative length prefix on a skipped bytes/string field would seek + # backwards; it must be rejected rather than corrupting the position. + prefix = self._encode_length_prefix(-5) + with io.BytesIO(prefix + b"payload") as bio: + decoder = avro.io.BinaryDecoder(bio) + self.assertRaises(avro.errors.InvalidAvroBinaryEncoding, decoder.skip_bytes) + + def test_skip_bytes_rejects_length_beyond_stream(self) -> None: + # A skipped bytes/string field declaring far more bytes than remain is + # rejected on a seekable reader before seeking past EOF. + prefix = self._encode_length_prefix(100 * 1024 * 1024) + with io.BytesIO(prefix) as bio: + decoder = avro.io.BinaryDecoder(bio) + self.assertRaises(avro.errors.InvalidAvroBinaryEncoding, decoder.skip_bytes) + + def test_skip_bytes_within_stream_still_skips(self) -> None: + # A well-formed skipped value advances the position past its data. + buf = io.BytesIO() + avro.io.BinaryEncoder(buf).write_bytes(b"hello") + trailer = b"AFTER" + with io.BytesIO(buf.getvalue() + trailer) as bio: + decoder = avro.io.BinaryDecoder(bio) + decoder.skip_bytes() + self.assertEqual(bio.read(), trailer) + + def test_skip_rejects_negative(self) -> None: + # The low-level skip() backstop rejects a negative byte count. + with io.BytesIO(b"abcdef") as bio: + decoder = avro.io.BinaryDecoder(bio) + self.assertRaises(avro.errors.InvalidAvroBinaryEncoding, decoder.skip, -1) + + +class TestDatumReaderCollectionAvailableBytes(unittest.TestCase): + """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 such as + ``null`` are not falsely rejected).""" + + @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)) + + def test_array_rejects_count_beyond_stream(self) -> None: + # A long array block count (1,000,000 long elements) with no element + # data following it. + buf = io.BytesIO() + avro.io.BinaryEncoder(buf).write_long(1_000_000) + self.assertRaises( + avro.errors.InvalidAvroBinaryEncoding, + self._decode, + '{"type": "array", "items": "long"}', + buf.getvalue(), + ) + + def test_map_rejects_count_beyond_stream(self) -> None: + buf = io.BytesIO() + avro.io.BinaryEncoder(buf).write_long(1_000_000) + self.assertRaises( + avro.errors.InvalidAvroBinaryEncoding, + self._decode, + '{"type": "map", "values": "long"}', + buf.getvalue(), + ) + + def test_array_of_null_not_falsely_rejected(self) -> None: + # null elements occupy zero bytes, so a large count is legitimate and + # must not be rejected. + count = 100_000 + buf = io.BytesIO() + enc = avro.io.BinaryEncoder(buf) + enc.write_long(count) # one block of `count` nulls + enc.write_long(0) # end-of-array marker + result = self._decode('{"type": "array", "items": "null"}', buf.getvalue()) + self.assertEqual(result, [None] * count) + + def test_small_truncated_array_still_rejected(self) -> None: + # A block count at or below _MAX_UNCHECKED_COLLECTION skips the + # bytes_remaining() pre-check (a perf optimization), but the elements + # still have to be read: a truncated small collection must not slip + # through. Here 10 longs are declared with no element bytes following. + self.assertLessEqual(10, avro.io._MAX_UNCHECKED_COLLECTION) + buf = io.BytesIO() + avro.io.BinaryEncoder(buf).write_long(10) + self.assertRaises( + avro.errors.InvalidAvroBinaryEncoding, + self._decode, + '{"type": "array", "items": "long"}', + buf.getvalue(), + ) + + def test_array_within_stream_still_reads(self) -> None: + schema_json = '{"type": "array", "items": "long"}' + schema = avro.schema.parse(schema_json) + buf = io.BytesIO() + writer = avro.io.DatumWriter(schema) + writer.write([1, 2, 3], avro.io.BinaryEncoder(buf)) + self.assertEqual(self._decode(schema_json, buf.getvalue()), [1, 2, 3]) + + def test_recursive_record_min_bytes_is_non_zero(self) -> None: + # A recursive record reference is not a zero-byte value: returning 0 for + # it would wrongly treat recursive records as zero-byte elements and + # weaken the bytes-remaining precheck. The minimum must stay >= 1. + schema = avro.schema.parse( + json.dumps( + { + "type": "record", + "name": "Node", + "fields": [{"name": "next", "type": ["null", "Node"]}], + } + ) + ) + self.assertGreaterEqual(avro.io._min_bytes_per_element(schema), 1) + # A union that includes the recursive record must not be underestimated + # below the >= 1 byte branch index either. + union = avro.schema.parse( + json.dumps( + [ + "null", + { + "type": "record", + "name": "Node", + "fields": [{"name": "next", "type": ["null", "Node"]}], + }, + ] + ) + ) + self.assertGreaterEqual(avro.io._min_bytes_per_element(union), 1) + + class TestMisc(unittest.TestCase): def test_decimal_bytes_small_scale(self) -> None: """Avro should raise an AvroTypeException when attempting to write a decimal with a larger exponent than the schema's scale.""" @@ -598,6 +803,33 @@ def test_unknown_symbol(self) -> None: datum_reader = avro.io.DatumReader(writers_schema, readers_schema) self.assertRaises(avro.errors.SchemaResolutionException, datum_reader.read, decoder) + def test_union_index_out_of_range(self) -> None: + # A union branch index that is negative or >= the number of branches is + # malformed and must be rejected before indexing (a negative index would + # otherwise wrap in Python and silently select the wrong branch). + schema = avro.schema.parse(json.dumps(["null", "long"])) + datum_reader = avro.io.DatumReader(schema) + for encoded in (b"\x0a", b"\x01"): # zig-zag long 5, and -1 + decoder = avro.io.BinaryDecoder(io.BytesIO(encoded)) + self.assertRaises(avro.errors.SchemaResolutionException, datum_reader.read, decoder) + + def test_enum_index_out_of_range(self) -> None: + # An enum symbol index that is negative or >= the number of symbols is + # malformed and must be rejected before indexing. + schema = avro.schema.parse(json.dumps({"type": "enum", "name": "E", "symbols": ["A", "B"]})) + datum_reader = avro.io.DatumReader(schema) + for encoded in (b"\x12", b"\x01"): # zig-zag int 9, and -1 + decoder = avro.io.BinaryDecoder(io.BytesIO(encoded)) + self.assertRaises(avro.errors.SchemaResolutionException, datum_reader.read, decoder) + + def test_read_long_rejects_overlong_varint(self) -> None: + # A 64-bit value uses at most 10 bytes; an 11th continuation byte is + # malformed and must be rejected rather than accepted as an arbitrarily + # large integer. + encoded = b"\x80" * 10 + b"\x01" + decoder = avro.io.BinaryDecoder(io.BytesIO(encoded)) + self.assertRaises(avro.errors.InvalidAvroBinaryEncoding, decoder.read_long) + def test_no_default_value(self) -> None: writers_schema = LONG_RECORD_SCHEMA datum_to_write = LONG_RECORD_DATUM @@ -825,6 +1057,256 @@ def test_raises_error_for_future_enum_symbol_without_default(self) -> None: reader.read(decoder) +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_int64_min_block_count(self) -> None: + # INT64_MIN as a block count is the pathological negation case. Python + # 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. + payload = b"\xff\xff\xff\xff\xff\xff\xff\xff\xff\x01\x00" + self.assertRaises( + avro.errors.AvroCollectionSizeException, + self._decode, + '{"type": "array", "items": "null"}', + payload, + ) + + 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_map_duplicate_keys_counted_cumulatively(self) -> None: + # Two blocks of 600 pairs that repeat the SAME key: len(read_items) would + # be 1, so a separate decoded-pair counter is needed to reject 1200 > 1000. + buf = io.BytesIO() + enc = avro.io.BinaryEncoder(buf) + for _ in range(2): + enc.write_long(600) + for _ in range(600): + enc.write_utf8("k") # same key; value is null (zero bytes) + enc.write_long(0) + with unittest.mock.patch.dict(os.environ, {"AVRO_MAX_COLLECTION_ITEMS": "1000"}): + self.assertRaises( + avro.errors.AvroCollectionSizeException, + self._decode, + '{"type": "map", "values": "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 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_skip_array_of_null_negative_block_respects_limit(self) -> None: + # A negative (byte-sized) block count must also be bounded when skipping, + # so it cannot be used to bypass the collection cap during resolution. + 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, negative=True), + ) + + def test_skip_block_bytes_wraps_seek_error(self) -> None: + # A sized (negative-count) skip block seeks by the declared byte size. If + # the underlying reader rejects the seek, surface it as an Avro decoding + # error rather than leaking a raw OSError/ValueError/OverflowError, or an + # AttributeError/TypeError from a reader lacking seek()/tell(). + for exc in (OSError, ValueError, OverflowError, AttributeError, TypeError): + + class FailingSkipDecoder: + def bytes_remaining(self) -> None: + return None + + def skip(self, n: int, _exc: type = exc) -> None: + raise _exc("cannot skip") + + self.assertRaises( + avro.errors.InvalidAvroBinaryEncoding, + avro.io.DatumReader._skip_block_bytes, + cast(avro.io.BinaryDecoder, FailingSkipDecoder()), + 10, # block_size + 5, # block_count + 1, # min_bytes > 0 so the skip is actually attempted (and fails) + ) + + def test_skip_block_bytes_rejects_positive_size_for_zero_byte_element(self) -> None: + # A zero-byte element type (min_bytes == 0) encodes to exactly 0 bytes, so + # its sized block must be empty; a positive block size would skip into the + # following fields and corrupt decoder alignment. + class NoSkipDecoder: + def bytes_remaining(self) -> None: + return None + + def skip(self, n: int) -> None: + if n != 0: # pragma: no cover - must not be reached + raise AssertionError("skip should not be called with a positive size for a zero-byte block") + + self.assertRaises( + avro.errors.InvalidAvroBinaryEncoding, + avro.io.DatumReader._skip_block_bytes, + cast(avro.io.BinaryDecoder, NoSkipDecoder()), + 10, # block_size (must be 0 for a zero-byte element type) + 5, # block_count + 0, # min_bytes (zero-byte element type) + ) + # A zero-size block for a zero-byte element type is valid and skips nothing. + avro.io.DatumReader._skip_block_bytes(cast(avro.io.BinaryDecoder, NoSkipDecoder()), 0, 5, 0) + + 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)) + # Assert the unset behavior inside a patch that snapshots os.environ, so + # the test never mutates the real process environment. + with unittest.mock.patch.dict(os.environ): + 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), + ) + + def test_non_zero_collection_bounded_by_structural_cap_when_unseekable(self) -> None: + # On a non-seekable reader the bytes-remaining check cannot run, so a huge + # non-zero-byte collection is bounded by the structural cap instead. + class NoTell: + def __init__(self, data: bytes) -> None: + self._b = io.BytesIO(data) + + def read(self, n: int = -1) -> bytes: + return self._b.read(n) + + schema = avro.schema.parse('{"type": "array", "items": "long"}') + reader = avro.io.DatumReader(schema) + with unittest.mock.patch.dict(os.environ, {"AVRO_MAX_COLLECTION_ITEMS": "1000"}): + decoder = avro.io.BinaryDecoder(cast(BinaryIO, NoTell(self._array_block(2000)))) + self.assertIsNone(decoder.bytes_remaining()) + self.assertRaises(avro.errors.AvroCollectionSizeException, reader.read, decoder) + + def load_tests(loader: unittest.TestLoader, default_tests: None, pattern: None) -> unittest.TestSuite: """Generate test cases across many test schema.""" suite = unittest.TestSuite() @@ -839,6 +1321,9 @@ def load_tests(loader: unittest.TestLoader, default_tests: None, pattern: None) ) suite.addTests(DefaultValueTestCase(field_type, default) for field_type, default in DEFAULT_VALUE_EXAMPLES) suite.addTests(loader.loadTestsFromTestCase(TestIncompatibleSchemaReading)) + suite.addTests(loader.loadTestsFromTestCase(TestBinaryDecoderAvailableBytes)) + suite.addTests(loader.loadTestsFromTestCase(TestDatumReaderCollectionAvailableBytes)) + suite.addTests(loader.loadTestsFromTestCase(TestDatumReaderCollectionSizeLimit)) return suite