Harden Cython bindings against buffer and lifetime misuse

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 = <ctamp.TampCompressor*>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 = <unsigned char *>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 = <unsigned char *> 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,
             <unsigned char *> buffer,

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 = <unsigned char *>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 <size_t>len(buf) <= written:
+                break
+            size = min(size, <size_t>len(buf) - written)
+            output_buffer_ptr = <unsigned char *>buf + written
             # decompress in chunks so we can still check PyErr_CheckSignals
             output_size = min(<size_t>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:
         """

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()