gh-156939: Detect buffer overflow in PyBytesWriter in debug mode

Lib/test/test_capi/test_bytes.py

@@ -1,5 +1,8 @@
+import textwrap
 import unittest
+from test import support
 from test.support import import_helper
+from test.support.script_helper import assert_python_failure
 
 _testlimitedcapi = import_helper.import_module('_testlimitedcapi')
 _testcapi = import_helper.import_module('_testcapi')
@@ -430,6 +433,28 @@ def test_example_resize(self):
     def test_example_highlevel(self):
         self.assertEqual(_testcapi.byteswriter_highlevel(), b'Hello World!')
 
+    @unittest.skipUnless(support.Py_DEBUG, 'need a Python debug build')
+    def test_canary_byte(self):
+        small_buffer = _testcapi.PyBytesWriter_small_buffer
+        large_size = small_buffer * 10
+
+        # Test small buffer and large buffer
+        for size in (0, 3, large_size):
+            with self.subTest(size=size):
+                code = textwrap.dedent(f"""
+                    from test.support import SuppressCrashReport
+                    import _testcapi
+                    size = {size}
+                    data = b'x' * size
+                    with SuppressCrashReport():
+                        _testcapi.byteswriter_test_canary_byte(data)
+                """)
+                proc = assert_python_failure('-c', code)
+                self.assertIn(b'Buffer overflow detected in PyBytesWriter',
+                              proc.err)
+                self.assertIn(f'at position {size}'.encode(),
+                              proc.err)
+
 
 class ByteArrayWriterTest(BaseWriterTest, unittest.TestCase):
     result_type = bytearray

Misc/NEWS.d/next/C_API/2026-09-04-16-41-07.gh-issue-156939.bKaQuE.rst

@@ -0,0 +1,2 @@
+When Python is built in debug mode, :c:type:`PyBytesWriter` now detects
+buffer overflow.  Patch by Victor Stinner.

Modules/_testcapi/bytes.c

@@ -151,7 +151,7 @@ writer_write_bytes(PyObject *self_raw, PyObject *args)
         return NULL;
     }
 
-    char *bytes;
+    const char *bytes;
     Py_ssize_t unused_size, size;
     if (!PyArg_ParseTuple(args, "y#n", &bytes, &unused_size, &size)) {
         return NULL;
@@ -454,13 +454,41 @@ test_byteswriter_ptr(PyObject *Py_UNUSED(module), PyObject *Py_UNUSED(args))
 }
 
 
+// Trigger a buffer overflow on purpose to test the canary byte feature
+// which detects buffer overflow
+static PyObject *
+byteswriter_test_canary_byte(PyObject *Py_UNUSED(module), PyObject *args)
+{
+    const char *str;
+    Py_ssize_t len;
+    if (!PyArg_ParseTuple(args, "s#", &str, &len)) {
+        return NULL;
+    }
+
+    PyBytesWriter *writer = PyBytesWriter_Create(len);
+    if (writer == NULL) {
+        return NULL;
+    }
+
+    char *data = PyBytesWriter_GetData(writer);
+    if (len) {
+        memcpy(data, str, len);
+    }
+    data[len] = '#';  // Overflow!
+
+    // In debug mode, PyBytesWriter_Finish() checks for buffer overflow
+    return PyBytesWriter_Finish(writer);
+}
+
+
 static PyMethodDef test_methods[] = {
     {"bytes_resize", bytes_resize, METH_VARARGS},
     {"bytes_join", bytes_join, METH_VARARGS},
     {"byteswriter_abc", byteswriter_abc, METH_NOARGS},
     {"byteswriter_resize", byteswriter_resize, METH_NOARGS},
     {"byteswriter_highlevel", byteswriter_highlevel, METH_NOARGS},
     {"test_byteswriter_ptr", test_byteswriter_ptr, METH_NOARGS},
+    {"byteswriter_test_canary_byte", byteswriter_test_canary_byte, METH_VARARGS},
     {NULL},
 };
 

Modules/fcntlmodule.c

@@ -121,13 +121,14 @@ fcntl_fcntl_impl(PyObject *module, int fd, int code, PyObject *arg)
             return PyBytes_FromStringAndSize(buf, len);
         }
         else {
-            PyBytesWriter *writer = PyBytesWriter_Create(len);
+            PyBytesWriter *writer = PyBytesWriter_Create(len + GUARDSZ);
             if (writer == NULL) {
                 PyBuffer_Release(&view);
                 return NULL;
             }
             char *ptr = PyBytesWriter_GetData(writer);
             memcpy(ptr, view.buf, len);
+            memcpy(ptr + len, guard, GUARDSZ);
             PyBuffer_Release(&view);
 
             do {
@@ -142,7 +143,7 @@ fcntl_fcntl_impl(PyObject *module, int fd, int code, PyObject *arg)
                 PyBytesWriter_Discard(writer);
                 return NULL;
             }
-            if (ptr[len] != '\0') {
+            if (memcmp(ptr + len, guard, GUARDSZ) != 0) {
                 PyErr_SetString(PyExc_SystemError,
                         "Memory corruption in fcntl() due to "
                         "buffer overflow. "
@@ -151,7 +152,8 @@ fcntl_fcntl_impl(PyObject *module, int fd, int code, PyObject *arg)
                 PyBytesWriter_Discard(writer);
                 return NULL;
             }
-            return PyBytesWriter_Finish(writer);
+            // Truncate the trailing guard bytes
+            return PyBytesWriter_FinishWithSize(writer, len);
         }
 #undef FCNTL_BUFSZ
     }
@@ -316,13 +318,14 @@ fcntl_ioctl_impl(PyObject *module, int fd, unsigned long code, PyObject *arg,
             return PyBytes_FromStringAndSize(buf, len);
         }
         else {
-            PyBytesWriter *writer = PyBytesWriter_Create(len);
+            PyBytesWriter *writer = PyBytesWriter_Create(len + GUARDSZ);
             if (writer == NULL) {
                 PyBuffer_Release(&view);
                 return NULL;
             }
             char *ptr = PyBytesWriter_GetData(writer);
             memcpy(ptr, view.buf, len);
+            memcpy(ptr + len, guard, GUARDSZ);
             PyBuffer_Release(&view);
 
             do {
@@ -337,7 +340,7 @@ fcntl_ioctl_impl(PyObject *module, int fd, unsigned long code, PyObject *arg,
                 PyBytesWriter_Discard(writer);
                 return NULL;
             }
-            if (ptr[len] != '\0') {
+            if (memcmp(ptr + len, guard, GUARDSZ) != 0) {
                 PyErr_SetString(PyExc_SystemError,
                         "Memory corruption in ioctl() due to "
                         "buffer overflow. "
@@ -346,7 +349,8 @@ fcntl_ioctl_impl(PyObject *module, int fd, unsigned long code, PyObject *arg,
                 PyBytesWriter_Discard(writer);
                 return NULL;
             }
-            return PyBytesWriter_Finish(writer);
+            // Truncate the trailing guard bytes
+            return PyBytesWriter_FinishWithSize(writer, len);
         }
 #undef IOCTL_BUFSZ
     }

Objects/bytesobject.c

@@ -3593,6 +3593,10 @@ _PyBytes_RepeatBuffer(char* dest, Py_ssize_t len_dest,
 
 // --- PyBytesWriter API -----------------------------------------------------
 
+// Use a value different than NUL (0) to be able to detect overflow writing
+// one extra NUL byte which is a common error.
+#define PyBytesWriter_CANARY_BYTE PYMEM_DEADBYTE
+
 static inline char*
 byteswriter_data(PyBytesWriter *writer)
 {
@@ -3604,7 +3608,8 @@ static inline Py_ssize_t
 byteswriter_allocated(PyBytesWriter *writer)
 {
     if (writer->obj == NULL) {
-        return sizeof(writer->small_buffer);
+        // Reserve the last byte for the canary byte
+        return sizeof(writer->small_buffer) - 1;
     }
     else if (writer->use_bytearray) {
         return PyByteArray_GET_SIZE(writer->obj);
@@ -3615,6 +3620,31 @@ byteswriter_allocated(PyBytesWriter *writer)
 }
 
 
+#ifdef Py_DEBUG
+static void
+byteswriter_check_canary_byte(PyBytesWriter *writer)
+{
+    const unsigned char *data = (const unsigned char*)byteswriter_data(writer);
+    unsigned char canary = data[writer->size];
+    if (canary != PyBytesWriter_CANARY_BYTE) {
+        _Py_FatalErrorFormat(__func__,
+                             "Buffer overflow detected in PyBytesWriter %p: "
+                             "one byte written after the buffer "
+                             "(at position %zd)",
+                             writer, writer->size);
+    }
+}
+
+
+static void
+byteswriter_write_canary_byte(PyBytesWriter *writer)
+{
+    unsigned char *data = (unsigned char*)byteswriter_data(writer);
+    data[writer->size] = PyBytesWriter_CANARY_BYTE;
+}
+#endif
+
+
 #ifdef MS_WINDOWS
    /* On Windows, overallocate by 50% is the best factor */
 #  define OVERALLOCATE_FACTOR 2
@@ -3719,6 +3749,7 @@ byteswriter_create(Py_ssize_t size, int use_bytearray)
     }
 #ifdef Py_DEBUG
     memset(byteswriter_data(writer), 0xff, byteswriter_allocated(writer));
+    byteswriter_write_canary_byte(writer);
 #endif
     return writer;
 }
@@ -3764,6 +3795,19 @@ PyBytesWriter_FinishWithSize(PyBytesWriter *writer, Py_ssize_t size)
         goto error;
     }
 
+#ifdef Py_DEBUG
+    // Check for buffer overflow
+    byteswriter_check_canary_byte(writer);
+
+    if (writer->obj != NULL) {
+        // byteswriter_write_canary_byte() can override the trailing NUL byte.
+        // So reset the trailing NUL byte to NUL.
+        Py_ssize_t allocated = byteswriter_allocated(writer);
+        char *data = byteswriter_data(writer);
+        data[allocated] = '\0';
+    }
+#endif
+
     PyObject *result;
     if (size == 0) {
         result = bytes_get_empty();
@@ -3850,6 +3894,9 @@ PyBytesWriter_Resize(PyBytesWriter *writer, Py_ssize_t size)
         return -1;
     }
     writer->size = size;
+#ifdef Py_DEBUG
+    byteswriter_write_canary_byte(writer);
+#endif
     return 0;
 }
 
@@ -3883,6 +3930,9 @@ PyBytesWriter_Grow(PyBytesWriter *writer, Py_ssize_t size)
         return -1;
     }
     writer->size = size;
+#ifdef Py_DEBUG
+    byteswriter_write_canary_byte(writer);
+#endif
     return 0;
 }
 
@@ -3948,5 +3998,8 @@ _PyBytesWriter_ResizeToAllocated(PyBytesWriter *writer)
 {
     Py_ssize_t allocated = byteswriter_allocated(writer);
     writer->size = allocated;
+#ifdef Py_DEBUG
+    byteswriter_write_canary_byte(writer);
+#endif
     return allocated;
 }