diff --git a/tamp/_c_compressor.pyx b/tamp/_c_compressor.pyx index 142a2999..979576f5 100644 --- a/tamp/_c_compressor.pyx +++ b/tamp/_c_compressor.pyx @@ -17,6 +17,21 @@ cdef class Compressor: cdef object f cdef bint _close_f_on_close cdef bint _dictionary_reset + cdef object _window_view # buffer export: blocks resizing the window bytearray under C + cdef bint _initialized + cdef bint _busy # set while C runs with the GIL released + + cdef int _check_ready(self) except -1: + if not self._initialized: + raise RuntimeError("Compressor is not initialized.") + if self._busy: + raise RuntimeError("Compressor is already in use by another thread.") + return 0 + + cdef int _enter(self) except -1: + self._check_ready() + self._busy = True + return 0 def __cinit__(self): self._c_compressor = PyMem_Malloc(sizeof(ctamp.TampCompressor)) @@ -46,6 +61,16 @@ cdef class Compressor: ): cdef ctamp.TampConf conf + if self._busy: + raise RuntimeError("Compressor is already in use by another thread.") + self._initialized = False + + # TampConf.window/literal are 4-bit bitfields; reject instead of wrapping. + if not 8 <= window <= 15: + raise ValueError("window must be in range [8, 15].") + if not 5 <= literal <= 8: + raise ValueError("literal must be in range [5, 8].") + if dictionary is not None and len(dictionary) != (1 << window): # The exact size is required: a too-small buffer used as the window # would let the C compressor write out of bounds. @@ -70,6 +95,7 @@ cdef class Compressor: conf.append = append self._window_buffer = dictionary if dictionary is not None else bytearray(1 << window) + self._window_view = memoryview(self._window_buffer) self._window_buffer_ptr = self._window_buffer self._dictionary_reset = dictionary_reset @@ -85,6 +111,7 @@ cdef class Compressor: if self._match_index is NULL: raise MemoryError self._attach_match_index() + self._initialized = True def write(self, const unsigned char[::1] data not None) -> int: cdef: @@ -110,16 +137,20 @@ cdef class Compressor: output_buffer_mv = memoryview(output_buffer) while input_remaining_size: - with nogil: - res = ctamp.tamp_compressor_compress( - self._c_compressor, - output_buffer_ptr, - CHUNK_SIZE, - &output_buffer_written_size, - data_ptr, - input_remaining_size, - &input_consumed_size - ) + self._enter() + try: + with nogil: + res = ctamp.tamp_compressor_compress( + self._c_compressor, + output_buffer_ptr, + CHUNK_SIZE, + &output_buffer_written_size, + data_ptr, + input_remaining_size, + &input_consumed_size + ) + finally: + self._busy = False if res < 0: raise ERROR_LOOKUP.get(res, NotImplementedError) self.f.write(output_buffer_mv[:output_buffer_written_size]) @@ -139,14 +170,18 @@ cdef class Compressor: cdef unsigned char *buffer_ptr = buffer cdef size_t buffer_len = len(buffer) - with nogil: - res = ctamp.tamp_compressor_flush( - self._c_compressor, - buffer_ptr, - buffer_len, - &output_written_size, - write_token, - ) + self._enter() + try: + with nogil: + res = ctamp.tamp_compressor_flush( + self._c_compressor, + buffer_ptr, + buffer_len, + &output_written_size, + write_token, + ) + finally: + self._busy = False if res < 0: raise ERROR_LOOKUP.get(res, NotImplementedError) @@ -163,6 +198,7 @@ cdef class Compressor: cdef bytearray buffer = bytearray(32) # Worst case: 23 (flush) + 4 (2x FLUSH) = 27 bytes cdef size_t output_written_size = 0 + self._check_ready() # runs with the GIL held res = ctamp.tamp_compressor_reset_dictionary( self._c_compressor, buffer, diff --git a/tamp/_c_decompressor.pyx b/tamp/_c_decompressor.pyx index 75987242..857eaf97 100644 --- a/tamp/_c_decompressor.pyx +++ b/tamp/_c_decompressor.pyx @@ -15,6 +15,10 @@ cdef class Decompressor: bytearray _window_buffer unsigned char *_window_buffer_ptr object input_buffer + object input_view # passed to f.readinto(); a view can't be resized + object _window_view # buffer export: blocks resizing the window bytearray under C + bint _initialized + bint _busy unsigned char *input_buffer_ptr size_t input_size size_t input_consumed @@ -34,6 +38,10 @@ cdef class Decompressor: ctamp.TampConf conf size_t input_consumed_size; + if self._busy: + raise RuntimeError("Decompressor is already in use by another thread.") + self._initialized = False + if not hasattr(f, "read"): # It's probably a path-like object. f = open(str(f), "rb") self._close_f_on_close = True @@ -43,6 +51,7 @@ cdef class Decompressor: self.f = f self.input_buffer = bytearray(CHUNK_SIZE) + self.input_view = memoryview(self.input_buffer) self.input_buffer_ptr = self.input_buffer self.input_size = 0 self.input_consumed = 0 @@ -70,48 +79,69 @@ cdef class Decompressor: raise ValueError("Dictionary-window size mismatch.") self._window_buffer = dictionary if dictionary is not None else bytearray(1 << conf.window) + self._window_view = memoryview(self._window_buffer) self._window_buffer_ptr = self._window_buffer res = ctamp.tamp_decompressor_init(self._c_decompressor, &conf, self._window_buffer_ptr, conf.window) if res < 0: raise ERROR_LOOKUP.get(res, NotImplementedError) + self._initialized = True - def readinto(self, bytearray buf) -> int: + def readinto(self, bytearray buf not None) -> int: cdef: - unsigned char *output_buffer_ptr = buf + unsigned char *output_buffer_ptr size_t output_written_size, input_chunk_consumed unsigned char *input_ptr size_t input_avail size_t output_size size_t size + size_t written = 0 ctamp.tamp_res res + if not self._initialized: + raise RuntimeError("Decompressor is not initialized.") + if self._busy: + raise RuntimeError("Decompressor is already in use by another thread.") size = len(buf) while size: + # f.readinto() below runs arbitrary Python that may resize buf, so re-derive + # the output pointer and clamp to its current length every iteration. + if len(buf) <= written: + break + size = min(size, len(buf) - written) + output_buffer_ptr = buf + written # decompress in chunks so we can still check PyErr_CheckSignals output_size = min(CHUNK_SIZE, size) input_ptr = self.input_buffer_ptr + self.input_consumed input_avail = self.input_size - with nogil: - res = ctamp.tamp_decompressor_decompress( - self._c_decompressor, - output_buffer_ptr, - output_size, - &output_written_size, - input_ptr, - input_avail, - &input_chunk_consumed - ) + self._busy = True + try: + with nogil: + res = ctamp.tamp_decompressor_decompress( + self._c_decompressor, + output_buffer_ptr, + output_size, + &output_written_size, + input_ptr, + input_avail, + &input_chunk_consumed + ) + finally: + self._busy = False self.input_size -= input_chunk_consumed self.input_consumed += input_chunk_consumed - output_buffer_ptr += output_written_size + written += output_written_size size -= output_written_size if res == ctamp.TAMP_INPUT_EXHAUSTED: # Read in more data if hasattr(self.f, "readinto"): - self.input_size = self.f.readinto(self.input_buffer) + n = self.f.readinto(self.input_view) + if n is None or not 0 <= n <= CHUNK_SIZE: + # C would read past input_buffer. + raise ValueError(f"readinto() returned invalid byte count {n!r}.") + self.input_size = n else: # Fall back for file-like objects that only implement read(). chunk = self.f.read(CHUNK_SIZE) @@ -130,7 +160,7 @@ cdef class Decompressor: # Check signals for things like KeyboardInterrupt PyErr_CheckSignals() - return len(buf) - size + return written def read(self, int size = -1) -> bytearray: """ diff --git a/tests/test_bug_regressions.py b/tests/test_bug_regressions.py index 0737bd9e..7c147540 100644 --- a/tests/test_bug_regressions.py +++ b/tests/test_bug_regressions.py @@ -236,5 +236,122 @@ def test_live_compression_roundtrip(self): self.assertEqual(bytes(d.read()), self.PAYLOAD) +def _c_extension_available(): + return micropython is None and CCompressor is not None + + +class _Raw(io.RawIOBase): + """Readable raw stream whose readinto() a test can override.""" + + def __init__(self, data): + self.b = io.BytesIO(data) + + def readable(self): + return True + + def read(self, n=-1): # RawIOBase.read() itself trusts readinto() + return self.b.read(n) + + +@unittest.skipUnless(_c_extension_available(), "Cython extension not built") +class TestCythonBufferLifetimes(unittest.TestCase): + """The Cython bindings hand raw pointers to C; Python-side misuse must not invalidate them.""" + + def setUp(self): + with io.BytesIO() as f: + c = CCompressor(f) + c.write(b"hello world " * 50) + c.flush() + self.compressed = f.getvalue() + + def test_dictionary_resize_blocked_while_in_use(self): + # Bug: resizing the dictionary bytearray freed storage that C still used. + d = bytearray(1 << 10) + c = CCompressor(io.BytesIO(), window=10, dictionary=d) # noqa: F841 (keep alive) + with self.assertRaises(BufferError): + d.clear() + + dd = bytearray(1 << 10) + dec = CDecompressor(io.BytesIO(self.compressed), dictionary=dd) # noqa: F841 (keep alive) + with self.assertRaises(BufferError): + dd.extend(b"x") + + def test_readinto_overreport_rejected(self): + # Bug: a readinto() returning more than the buffer size made C read past it. + class Overreporting(_Raw): + def readinto(self, buf): + n = self.b.readinto(buf) + return n and (1 << 28) + + with self.assertRaisesRegex(ValueError, "readinto"): + CDecompressor(Overreporting(self.compressed)).read() + + def test_readinto_cannot_resize_input_buffer(self): + # Bug: readinto() received the internal bytearray and could shrink it under C. + class Shrinking(_Raw): + def readinto(self, buf): + chunk = self.b.read(len(buf)) + buf[:] = chunk + return len(chunk) + + with self.assertRaises(ValueError): + CDecompressor(Shrinking(self.compressed)).read() + + def test_readinto_output_buffer_resized_by_source(self): + # Bug: readinto() cached the output pointer across f.readinto(), which may run code + # that resizes the caller's buffer. + out = bytearray(1 << 16) + + class ShrinksOutput(_Raw): + def readinto(self, buf): + del out[16:] + return self.b.readinto(buf) + + n = CDecompressor(ShrinksOutput(self.compressed)).readinto(out) + self.assertLessEqual(n, len(out)) + self.assertEqual(bytes(out[:n]), (b"hello world " * 50)[:n]) + + def test_methods_before_init_raise(self): + # Bug: __new__ without __init__ ran C on an uninitialized struct. + c = CCompressor.__new__(CCompressor) + with self.assertRaises(RuntimeError): + c.write(b"abc") + with self.assertRaises(RuntimeError): + c.flush() + d = CDecompressor.__new__(CDecompressor) + with self.assertRaises(RuntimeError): + d.readinto(bytearray(10)) + + def test_window_literal_out_of_range(self): + # Bug: window/literal wrapped into 4-bit bitfields (window=24 -> 8). + for kw in ({"window": 24}, {"window": 30}, {"literal": 24}, {"literal": 21}): + with self.subTest(**kw), self.assertRaises(ValueError): + CCompressor(io.BytesIO(), **kw) + + def test_concurrent_write_rejected(self): + # Bug: 2 threads inside the GIL-released compress loop on one Compressor crashed. + import threading + + c = CCompressor(io.BytesIO(), window=10) + blob = bytes(range(256)) * 4096 + errors = [] + start = threading.Barrier(4) + + def worker(): + start.wait() + try: + for _ in range(10): + c.write(blob) + except RuntimeError as e: + errors.append(e) + + threads = [threading.Thread(target=worker) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertTrue(errors) + + if __name__ == "__main__": unittest.main()