This is an automated email from the ASF dual-hosted git repository.
RyanSkraba pushed a commit to branch branch-1.12
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/branch-1.12 by this push:
new 478ed0b433 AVRO-4290: [python] Enforce a maximum decompressed block
size (#3850)
478ed0b433 is described below
commit 478ed0b433d9bd1db3a2781f8b6b351b9b39d735
Author: Ismaël Mejía <[email protected]>
AuthorDate: Fri Aug 7 09:14:14 2026 +0200
AVRO-4290: [python] Enforce a maximum decompressed block size (#3850)
* AVRO-4290: [python] Enforce a maximum decompressed block size
When reading a data file, each block is decompressed according to the
file's codec. A block with a very high compression ratio (or a malformed
block) could expand to far more memory than its compressed size. Enforce a
configurable maximum decompressed size across the deflate, bzip2, snappy and
zstandard codecs, mirroring the Java SDK's decompression limit (AVRO-4247).
The limit defaults to 200 MiB and can be overridden with the
AVRO_MAX_DECOMPRESS_LENGTH environment variable; exceeding it raises
AvroDecompressionSizeException.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Address review: avoid deflate copy; zstd
check-before-extend
- DeflateCodec.decompress accumulates into a bytearray so the flush()
output is
appended in place instead of creating an extra full-size copy of the
already
decompressed data.
- ZstandardCodec.decompress checks len(uncompressed) + len(chunk) before
extending, so the buffer never grows past the configured limit.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Verify bzip2 stream EOF and drain fully
BZip2Codec.decompress now loops the BZ2Decompressor: it drains all buffered
output (bounded by the limit), verifies the stream reached EOF (rejecting a
truncated/corrupt block with InvalidAvroBinaryEncoding), and handles
concatenated bzip2 streams as bz2.decompress does. Add a truncated-block
test.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Reject truncated deflate and short snappy blocks
Address review feedback:
- DeflateCodec: after flushing the decompressor, verify decompressor.eof so
a
truncated/incomplete raw-deflate block is rejected with
InvalidAvroBinaryEncoding (zlib.decompress used to raise for this; the
decompressobj-based size cap otherwise silently accepted partial output).
- SnappyCodec: validate the block length is >= 4 before reading length - 4
bytes, raising a codec-specific InvalidAvroBinaryEncoding instead of
falling
through to the generic decoder error.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Add a truncated-deflate-block rejection test
Cover the new DeflateCodec behavior (raising InvalidAvroBinaryEncoding when
the
end-of-stream marker isn't reached) with a test that truncates a valid
deflate
block and asserts decompress() rejects it, mirroring the bzip2 truncation
test.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Bound the deflate flush() output too
The previous fix bounded decompress(data, limit+1) but then called
decompressor.flush() with no limit, so the residual output (from the
unconsumed input left by max_length) could still expand unbounded. Drain the
unconsumed_tail in a loop with a per-call max_length and pass a bounded
length
to flush(), so the accumulated output can never exceed the limit by more
than
one byte before being rejected.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Clamp AVRO_MAX_DECOMPRESS_LENGTH to sys.maxsize
_max_decompress_length() returned the raw parsed int. An absurdly large
override
is passed as max_length to zlib/bz2 decompress(), which raise OverflowError
when
it exceeds Py_ssize_t. Clamp to sys.maxsize so an oversized override is
honored
as "effectively unbounded" instead of producing a confusing OverflowError.
Assisted-by: GitHub Copilot:claude-opus-4.8
* AVRO-4290: [python] Avoid limit+1 Py_ssize_t overflow at sys.maxsize
After clamping AVRO_MAX_DECOMPRESS_LENGTH to sys.maxsize, the deflate/bz2
paths
still computed `limit + 1` for the decompress() max_length, which overflows
Py_ssize_t (and raises OverflowError) when limit == sys.maxsize. Add
_decompress_read_ceiling(), which returns limit + 1 normally but sys.maxsize
when the limit is already sys.maxsize (no realistic block can exceed it),
and
use it for the bounded read/flush in both paths.
Assisted-by: GitHub Copilot:claude-opus-4.8
---
lang/py/avro/codecs.py | 138 ++++++++++++++++++++++++++++++++++++++-
lang/py/avro/errors.py | 8 +++
lang/py/avro/test/test_codecs.py | 127 +++++++++++++++++++++++++++++++++++
3 files changed, 271 insertions(+), 2 deletions(-)
diff --git a/lang/py/avro/codecs.py b/lang/py/avro/codecs.py
index e1655fef43..6828483335 100644
--- a/lang/py/avro/codecs.py
+++ b/lang/py/avro/codecs.py
@@ -29,6 +29,8 @@ converting charsets
(https://docs.python.org/3/library/codecs.html).
import abc
import binascii
import io
+import os
+import sys
import struct
import zlib
from typing import Dict, Tuple, Type
@@ -41,6 +43,73 @@ import avro.io
#
STRUCT_CRC32 = struct.Struct(">I") # big-endian unsigned int
+# Name of the environment variable used to override the default maximum size of
+# a single decompressed data-file block.
+MAX_DECOMPRESS_LENGTH_ENV = "AVRO_MAX_DECOMPRESS_LENGTH"
+
+# Default upper bound, in bytes, on the size a single data-file block may
+# decompress to. A block with a very high compression ratio (or a malformed
+# block) can otherwise expand to far more memory than its compressed size.
+# Reading a block that would decompress beyond this limit raises an
+# :class:`avro.errors.AvroDecompressionSizeException`. This mirrors the Java
+# SDK's ``org.apache.avro.limits.decompress.maxLength`` limit (AVRO-4247). The
+# default may be overridden with the ``AVRO_MAX_DECOMPRESS_LENGTH`` environment
+# variable.
+DEFAULT_MAX_DECOMPRESS_LENGTH = 200 * 1024 * 1024 # 200 MiB
+
+
+def _max_decompress_length() -> int:
+ """Return the maximum decompressed block size, honoring the environment
override."""
+ value = os.environ.get(MAX_DECOMPRESS_LENGTH_ENV)
+ if value is None:
+ return DEFAULT_MAX_DECOMPRESS_LENGTH
+ try:
+ parsed = int(value)
+ except ValueError:
+ return DEFAULT_MAX_DECOMPRESS_LENGTH
+ if parsed <= 0:
+ return DEFAULT_MAX_DECOMPRESS_LENGTH
+ # Clamp to sys.maxsize so the value stays a valid Py_ssize_t: it is passed
as
+ # the max_length to zlib/bz2 decompress(), which raise OverflowError for a
+ # value that does not fit. sys.maxsize is already far larger than any real
+ # block, so clamping only affects absurd overrides.
+ return min(parsed, sys.maxsize)
+
+
+def _raise_decompression_too_large(limit: int) -> None:
+ raise avro.errors.AvroDecompressionSizeException(f"Decompressed block size
exceeds the maximum allowed of {limit} bytes")
+
+
+def _decompress_read_ceiling(limit: int) -> int:
+ """Return limit + 1 (one byte past the limit, to detect an over-limit
block),
+ but never exceeding sys.maxsize so the value stays a valid Py_ssize_t.
+
+ zlib/bz2 decompress() raise OverflowError for a max_length above
Py_ssize_t.
+ When the limit is already sys.maxsize (an oversized env override, clamped
by
+ _max_decompress_length), no realistic block can exceed it, so returning
+ sys.maxsize instead of sys.maxsize + 1 is safe.
+ """
+ return limit if limit >= sys.maxsize else limit + 1
+
+
+def _snappy_uncompressed_length(data: bytes) -> "int | None":
+ """Return the uncompressed length declared in a raw Snappy block header.
+
+ The Snappy format prefixes the compressed data with the uncompressed length
+ encoded as a little-endian base-128 varint. Returns ``None`` if the header
+ cannot be parsed.
+ """
+ result = 0
+ shift = 0
+ for byte in data:
+ result |= (byte & 0x7F) << shift
+ if not (byte & 0x80):
+ return result
+ shift += 7
+ if shift > 63:
+ break
+ return None
+
def _check_crc32(bytes_: bytes, checksum: bytes) -> None:
if binascii.crc32(bytes_) & 0xFFFFFFFF != STRUCT_CRC32.unpack(checksum)[0]:
@@ -123,7 +192,32 @@ class DeflateCodec(Codec):
data = readers_decoder.read_bytes()
# -15 is the log of the window size; negative indicates
# "raw" (no zlib headers) decompression. See zlib.h.
- uncompressed = zlib.decompress(data, -15)
+ limit = _max_decompress_length()
+ decompressor = zlib.decompressobj(-15)
+ # Decompress in bounded steps: request at most (limit + 1 - produced)
+ # bytes each call so the accumulated output can never exceed the limit
by
+ # more than one byte before being rejected. decompress()'s max_length
+ # leaves unconsumed input in unconsumed_tail, and flush() would
otherwise
+ # emit the remainder unbounded, so drain the tail in a loop and bound
the
+ # final flush too.
+ ceiling = _decompress_read_ceiling(limit)
+ uncompressed = bytearray()
+ pending = data
+ while True:
+ want = ceiling - len(uncompressed)
+ uncompressed += decompressor.decompress(pending, want)
+ if len(uncompressed) > limit:
+ _raise_decompression_too_large(limit)
+ pending = decompressor.unconsumed_tail
+ if not pending:
+ break
+ uncompressed += decompressor.flush(ceiling - len(uncompressed))
+ if len(uncompressed) > limit:
+ _raise_decompression_too_large(limit)
+ if not decompressor.eof:
+ # The end-of-stream marker was not reached: the block is truncated
or
+ # corrupt. zlib.decompress() used to raise for this; preserve that.
+ raise avro.errors.InvalidAvroBinaryEncoding("Truncated or corrupt
deflate block")
return avro.io.BinaryDecoder(io.BytesIO(uncompressed))
@@ -139,7 +233,31 @@ if has_bzip2:
def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
length = readers_decoder.read_long()
data = readers_decoder.read(length)
- uncompressed = bz2.decompress(data)
+ limit = _max_decompress_length()
+ ceiling = _decompress_read_ceiling(limit)
+ uncompressed = bytearray()
+ decompressor = bz2.BZ2Decompressor()
+ input_data = data
+ while True:
+ # Request enough to detect exceeding the limit without
allocating
+ # the full (potentially huge) output.
+ want = ceiling - len(uncompressed)
+ if want <= 0:
+ want = 1
+ uncompressed += decompressor.decompress(input_data, want)
+ input_data = b"" # subsequent calls drain buffered output
+ if len(uncompressed) > limit:
+ _raise_decompression_too_large(limit)
+ if decompressor.eof:
+ # Handle concatenated bzip2 streams as bz2.decompress does.
+ input_data = decompressor.unused_data
+ if not input_data:
+ break
+ decompressor = bz2.BZ2Decompressor()
+ elif decompressor.needs_input:
+ # All input consumed but the stream did not end: truncated
or corrupt.
+ raise avro.errors.InvalidAvroBinaryEncoding("Truncated or
corrupt bzip2 block")
+ # otherwise output was capped for this call; loop to drain more
return avro.io.BinaryDecoder(io.BytesIO(uncompressed))
@@ -157,8 +275,20 @@ if has_snappy:
def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
# Compressed data includes a 4-byte CRC32 checksum
length = readers_decoder.read_long()
+ if length < 4:
+ raise avro.errors.InvalidAvroBinaryEncoding(
+ f"Invalid snappy block length {length}: must be at least 4
bytes for the trailing CRC32 checksum"
+ )
data = readers_decoder.read(length - 4)
+ limit = _max_decompress_length()
+ # The Snappy block header declares the uncompressed length as a
+ # varint; reject an over-large block before allocating for it.
+ declared = _snappy_uncompressed_length(data)
+ if declared is not None and declared > limit:
+ _raise_decompression_too_large(limit)
uncompressed = snappy.decompress(data)
+ if len(uncompressed) > limit:
+ _raise_decompression_too_large(limit)
checksum = readers_decoder.read(4)
_check_crc32(uncompressed, checksum)
return avro.io.BinaryDecoder(io.BytesIO(uncompressed))
@@ -176,6 +306,7 @@ if has_zstandard:
def decompress(readers_decoder: avro.io.BinaryDecoder) ->
avro.io.BinaryDecoder:
length = readers_decoder.read_long()
data = readers_decoder.read(length)
+ limit = _max_decompress_length()
uncompressed = bytearray()
dctx = zstd.ZstdDecompressor()
with dctx.stream_reader(io.BytesIO(data)) as reader:
@@ -183,6 +314,9 @@ if has_zstandard:
chunk = reader.read(16384)
if not chunk:
break
+ # Check before extending so the buffer never grows past
the limit.
+ if len(uncompressed) + len(chunk) > limit:
+ _raise_decompression_too_large(limit)
uncompressed.extend(chunk)
return avro.io.BinaryDecoder(io.BytesIO(uncompressed))
diff --git a/lang/py/avro/errors.py b/lang/py/avro/errors.py
index 3939a364fa..fd0287c8dd 100644
--- a/lang/py/avro/errors.py
+++ b/lang/py/avro/errors.py
@@ -92,6 +92,14 @@ class AvroOutOfScaleException(AvroTypeException):
return super().__init__(f"The exponent of {datum}, {exponent}, is too
large for the schema scale of {scale}")
+class AvroDecompressionSizeException(AvroException):
+ """Raised when a data-file block decompresses to more than the configured
maximum.
+
+ This guards against unbounded memory allocation when a highly compressible
+ (or malformed) block would expand to far more than its compressed size.
+ """
+
+
class SchemaResolutionException(AvroException):
def __init__(self, fail_msg, writers_schema=None, readers_schema=None,
*args):
writers_message = f"\nWriter's Schema: {_safe_pretty(writers_schema)}"
if writers_schema else ""
diff --git a/lang/py/avro/test/test_codecs.py b/lang/py/avro/test/test_codecs.py
new file mode 100644
index 0000000000..5a462f0eca
--- /dev/null
+++ b/lang/py/avro/test/test_codecs.py
@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+
+##
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests that data-file block decompression enforces a maximum output size.
+
+A block with a very high compression ratio (or a malformed block) can expand to
+far more memory than its compressed size; these tests ensure that decompressing
+such a block raises an error instead of allocating without bound.
+"""
+
+import io
+import os
+import unittest
+from typing import Type
+
+import avro.codecs
+import avro.errors
+import avro.io
+
+
+class TestDecompressionSizeLimit(unittest.TestCase):
+ LIMIT = 1024
+
+ def setUp(self) -> None:
+ self._previous = os.environ.get(avro.codecs.MAX_DECOMPRESS_LENGTH_ENV)
+ os.environ[avro.codecs.MAX_DECOMPRESS_LENGTH_ENV] = str(self.LIMIT)
+
+ def tearDown(self) -> None:
+ if self._previous is None:
+ os.environ.pop(avro.codecs.MAX_DECOMPRESS_LENGTH_ENV, None)
+ else:
+ os.environ[avro.codecs.MAX_DECOMPRESS_LENGTH_ENV] = self._previous
+
+ def _decoder_for(self, codec: Type[avro.codecs.Codec], data: bytes) ->
avro.io.BinaryDecoder:
+ """Compress `data` with the codec and wrap it as a data-file block."""
+ compressed, _ = codec.compress(data)
+ buffer = io.BytesIO()
+ encoder = avro.io.BinaryEncoder(buffer)
+ encoder.write_long(len(compressed))
+ buffer.write(compressed)
+ return avro.io.BinaryDecoder(io.BytesIO(buffer.getvalue()))
+
+ def _assert_over_limit_rejected(self, codec: Type[avro.codecs.Codec]) ->
None:
+ # A block of zeros far larger than the limit but that compresses tiny.
+ decoder = self._decoder_for(codec, b"\x00" * (self.LIMIT * 8))
+ with self.assertRaises(avro.errors.AvroDecompressionSizeException):
+ codec.decompress(decoder)
+
+ def _assert_within_limit_ok(self, codec: Type[avro.codecs.Codec]) -> None:
+ payload = b"the quick brown fox " * 10 # well under the limit
+ decoder = self._decoder_for(codec, payload)
+ self.assertEqual(payload, codec.decompress(decoder).reader.read())
+
+ def test_deflate_over_limit(self) -> None:
+ self._assert_over_limit_rejected(avro.codecs.DeflateCodec)
+
+ def test_deflate_within_limit(self) -> None:
+ self._assert_within_limit_ok(avro.codecs.DeflateCodec)
+
+ def test_deflate_truncated_block_rejected(self) -> None:
+ """A truncated deflate block must be rejected rather than silently
accepted."""
+ compressed, _ = avro.codecs.DeflateCodec.compress(b"the quick brown
fox " * 10)
+ truncated = compressed[:-4] # drop the tail so the end-of-stream
marker is never reached
+ buffer = io.BytesIO()
+ encoder = avro.io.BinaryEncoder(buffer)
+ encoder.write_long(len(truncated))
+ buffer.write(truncated)
+ decoder = avro.io.BinaryDecoder(io.BytesIO(buffer.getvalue()))
+ with self.assertRaises(avro.errors.InvalidAvroBinaryEncoding):
+ avro.codecs.DeflateCodec.decompress(decoder)
+
+ @unittest.skipUnless(avro.codecs.has_bzip2, "bzip2 not available")
+ def test_bzip2_over_limit(self) -> None:
+ self._assert_over_limit_rejected(avro.codecs.BZip2Codec)
+
+ @unittest.skipUnless(avro.codecs.has_bzip2, "bzip2 not available")
+ def test_bzip2_within_limit(self) -> None:
+ self._assert_within_limit_ok(avro.codecs.BZip2Codec)
+
+ @unittest.skipUnless(avro.codecs.has_bzip2, "bzip2 not available")
+ def test_bzip2_truncated_block_rejected(self) -> None:
+ """A truncated bzip2 block must be rejected rather than silently
accepted."""
+ compressed, _ = avro.codecs.BZip2Codec.compress(b"the quick brown fox
" * 10)
+ truncated = compressed[:-4] # drop the tail so the stream never
reaches EOF
+ buffer = io.BytesIO()
+ encoder = avro.io.BinaryEncoder(buffer)
+ encoder.write_long(len(truncated))
+ buffer.write(truncated)
+ decoder = avro.io.BinaryDecoder(io.BytesIO(buffer.getvalue()))
+ with self.assertRaises(avro.errors.InvalidAvroBinaryEncoding):
+ avro.codecs.BZip2Codec.decompress(decoder)
+
+ @unittest.skipUnless(avro.codecs.has_snappy, "snappy not available")
+ def test_snappy_over_limit(self) -> None:
+ self._assert_over_limit_rejected(avro.codecs.SnappyCodec)
+
+ @unittest.skipUnless(avro.codecs.has_snappy, "snappy not available")
+ def test_snappy_within_limit(self) -> None:
+ self._assert_within_limit_ok(avro.codecs.SnappyCodec)
+
+ @unittest.skipUnless(avro.codecs.has_zstandard, "zstandard not available")
+ def test_zstandard_over_limit(self) -> None:
+ self._assert_over_limit_rejected(avro.codecs.ZstandardCodec)
+
+ @unittest.skipUnless(avro.codecs.has_zstandard, "zstandard not available")
+ def test_zstandard_within_limit(self) -> None:
+ self._assert_within_limit_ok(avro.codecs.ZstandardCodec)
+
+
+if __name__ == "__main__":
+ unittest.main()
