Reject trailing Brotli data to prevent infinite drain loops (#232)

Author: illia-v

HISTORY.rst

@@ -1,6 +1,16 @@
 Changelog
 =========
 
+Unreleased
+----------
+
+- Changed one-shot ``decompress()`` and streaming ``Decompressor.process()`` /
+  ``Decompressor.decompress()`` to raise ``brotlicffi.error`` for trailing bytes
+  after a complete Brotli stream, including calls without an output limit.
+  Previously tolerated input is now rejected, and any output produced during
+  the failing call is discarded. This prevents infinite drain loops in
+  streaming consumers.
+
 1.2.0.2 (2026-08-21)
 --------------------
 

src/brotlicffi/_api.py

@@ -492,6 +492,11 @@ def _decompress(self, data, output_buffer_limit):
                     b"Decompression error: %s" % ffi.string(error_message)
                 )
 
+            # Reject bytes remaining after the Brotli stream finishes.
+            if rc == lib.BROTLI_DECODER_RESULT_SUCCESS and available_in[0]:
+                self._unconsumed_data = b''
+                raise error("Decompression error: trailing data after stream.")
+
             # Next, copy the result out.
             chunk = ffi.buffer(out_buffer, buffer_size - available_out[0])[:]
             chunks.append(chunk)

test/test_simple_decompression.py

@@ -108,6 +108,45 @@ def test_high_expansion_prefix_without_output_buffer_limit():
     assert result == uncompressed
 
 
+def test_decompression_rejects_trailing_data():
+    data = brotlicffi.compress(b'A' * 100) + b'x'
+    with pytest.raises(brotlicffi.error, match='trailing data'):
+        brotlicffi.decompress(data)
+
+
+@pytest.mark.parametrize('method', ['decompress', 'process'])
+@pytest.mark.parametrize('output_buffer_limit', [None, 50])
+def test_decompressobj_rejects_trailing_data(method, output_buffer_limit):
+    o = brotlicffi.Decompressor()
+    decompress = getattr(o, method)
+    # Without a limit, decoding needs multiple internal buffer iterations.
+    data = brotlicffi.compress(b'A' * 100) + b'tail'
+    if output_buffer_limit is not None:
+        assert decompress(data, output_buffer_limit=50) == b'A' * 50
+        assert not o.can_accept_more_data()
+        data = b''
+
+    with pytest.raises(brotlicffi.error, match='trailing data'):
+        decompress(data, output_buffer_limit=output_buffer_limit)
+
+    assert o._unconsumed_data == b''
+    with pytest.raises(brotlicffi.error, match='trailing data'):
+        decompress(b'x', output_buffer_limit=output_buffer_limit)
+
+
+@pytest.mark.parametrize('output_buffer_limit', [None, 100])
+def test_decompressobj_rejects_data_after_finished(output_buffer_limit):
+    o = brotlicffi.Decompressor()
+    data = brotlicffi.compress(b'A' * 100)
+    assert (
+        o.process(data, output_buffer_limit=output_buffer_limit) == b'A' * 100
+    )
+    assert o.is_finished()
+
+    with pytest.raises(brotlicffi.error, match='trailing data'):
+        o.process(b'x', output_buffer_limit=output_buffer_limit)
+
+
 def test_drip_feed(simple_compressed_file):
     """
     Sending in the data one byte at a time still works.