mirrors/brotli
0
0
Fork 0
mirror of https://github.com/google/brotli.git synced 2024-11-23 09:57:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

Ensure decompression consumes all input (#730)

Browse Source 
* Ensure decompression consumes all input

If not, it's a corrupt stream.

* Use byte strings
This commit is contained in:
Justin Ridgewell 2018-11-12 04:36:00 -05:00 committed by Eugene Kliuchnikov
parent d0ffe60b87
commit 5805f99a53
3 changed files with 15 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
python/_brotli.cc
View File

@ -414,7 +414,7 @@ static BROTLI_BOOL decompress_stream(BrotliDecoderState* dec,
(*output).insert((*output).end(), buffer, buffer + buffer_length);
}
}
ok = result != BROTLI_DECODER_RESULT_ERROR;
ok = result != BROTLI_DECODER_RESULT_ERROR && !available_in;
Py_END_ALLOW_THREADS
return ok;
@ -672,7 +672,7 @@ static PyObject* brotli_decompress(PyObject *self, PyObject *args, PyObject *key
if (available_out != 0)
output.insert(output.end(), next_out, next_out + available_out);
}
ok = result == BROTLI_DECODER_RESULT_SUCCESS;
ok = result == BROTLI_DECODER_RESULT_SUCCESS && !available_in;
BrotliDecoderDestroyInstance(state);
Py_END_ALLOW_THREADS

4
python/tests/decompress_test.py
View File

@ -31,6 +31,10 @@ class TestDecompress(_test_utils.TestCase):
self._decompress(test_data)
self._check_decompression(test_data)
def test_garbage_appended(self):
with self.assertRaises(brotli.error):
brotli.decompress(brotli.compress(b'a') + b'a')
_test_utils.generate_test_methods(TestDecompress, for_decompression=True)

9
python/tests/decompressor_test.py
View File

@ -43,6 +43,15 @@ class TestDecompressor(_test_utils.TestCase):
self._decompress(test_data)
self._check_decompression(test_data)
def test_garbage_appended(self):
with self.assertRaises(brotli.error):
self.decompressor.process(brotli.compress(b'a') + b'a')
def test_already_finished(self):
self.decompressor.process(brotli.compress(b'a'))
with self.assertRaises(brotli.error):
self.decompressor.process(b'a')
_test_utils.generate_test_methods(TestDecompressor, for_decompression=True)