From 89a8ac91f07851c8cb554bb9a4b1126523616c44 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Wed, 9 Sep 2026 21:22:34 +0200
Subject: [PATCH 1/9] gh-156939: Detect buffer overflow in PyBytesWriter in
 debug mode

Reserve one byte in PyBytesWriter used as a canary byte: set it to a
special value. PyBytesWriter_Finish() checks if the canary byte has
been overriden. Add a test on the feature.

Update buffer overflow check in fcntl: allocate extra guard bytes in
the writer and then truncate these bytes.
---
 Lib/test/test_capi/test_bytes.py              | 21 +++++++
 ...-09-04-16-41-07.gh-issue-156939.bKaQuE.rst |  2 +
 Modules/_testcapi/bytes.c                     | 35 +++++++++++-
 Modules/fcntlmodule.c                         | 16 ++++--
 Objects/bytesobject.c                         | 55 ++++++++++++++++++-
 5 files changed, 121 insertions(+), 8 deletions(-)
 create mode 100644 Misc/NEWS.d/next/C_API/2026-09-04-16-41-07.gh-issue-156939.bKaQuE.rst

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index 4c1431bacef0a27..d997e578f635ea2 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -1,5 +1,7 @@
+import textwrap
 import unittest
 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 +432,25 @@ def test_example_resize(self):
     def test_example_highlevel(self):
         self.assertEqual(_testcapi.byteswriter_highlevel(), b'Hello World!')
 
+    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)
+
 
 class ByteArrayWriterTest(BaseWriterTest, unittest.TestCase):
     result_type = bytearray
diff --git a/Misc/NEWS.d/next/C_API/2026-09-04-16-41-07.gh-issue-156939.bKaQuE.rst b/Misc/NEWS.d/next/C_API/2026-09-04-16-41-07.gh-issue-156939.bKaQuE.rst
new file mode 100644
index 000000000000000..57c9a0ab46e9046
--- /dev/null
+++ b/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.
diff --git a/Modules/_testcapi/bytes.c b/Modules/_testcapi/bytes.c
index 4830cc8b54bd837..83d1dab6c749f94 100644
--- a/Modules/_testcapi/bytes.c
+++ b/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,6 +454,38 @@ 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(0);
+    if (writer == NULL) {
+        goto error;
+    }
+    if (PyBytesWriter_Grow(writer, len) < 0) {
+        goto error;
+    }
+    char *data = PyBytesWriter_GetData(writer);
+    if (len) {
+        memcpy(data, str, len);
+    }
+    data[len] = '#';  // Overflow!
+
+    return PyBytesWriter_Finish(writer);
+
+error:
+    PyBytesWriter_Discard(writer);
+    return NULL;
+}
+
+
 static PyMethodDef test_methods[] = {
     {"bytes_resize", bytes_resize, METH_VARARGS},
     {"bytes_join", bytes_join, METH_VARARGS},
@@ -461,6 +493,7 @@ static PyMethodDef test_methods[] = {
     {"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},
 };
 
diff --git a/Modules/fcntlmodule.c b/Modules/fcntlmodule.c
index e6a40ffc5a26144..5dd3df9bb408f0c 100644
--- a/Modules/fcntlmodule.c
+++ b/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
     }
diff --git a/Objects/bytesobject.c b/Objects/bytesobject.c
index 2ae55b33f4f49d7..a7b6b5a7df1e937 100644
--- a/Objects/bytesobject.c
+++ b/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;
 }

From 95e37016f1ac0792495b9a10ce413c271c9b6357 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Wed, 9 Sep 2026 21:47:03 +0200
Subject: [PATCH 2/9] Skip test_canary_byte on release build

---
 Lib/test/test_capi/test_bytes.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index d997e578f635ea2..632881eb1b940ed 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -1,5 +1,6 @@
 import textwrap
 import unittest
+from test import support
 from test.support import import_helper
 from test.support.script_helper import assert_python_failure
 
@@ -432,6 +433,7 @@ 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

From 0dbd2cc2855e901dcc8995d2829320509233b22a Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Wed, 9 Sep 2026 21:53:37 +0200
Subject: [PATCH 3/9] Cleanup test_canary_byte()

Avoid calling PyBytesWriter_Grow().
---
 Lib/test/test_capi/test_bytes.py |  2 ++
 Modules/_testcapi/bytes.c        | 13 ++++---------
 2 files changed, 6 insertions(+), 9 deletions(-)

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index 632881eb1b940ed..de53c9bc45670c2 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -452,6 +452,8 @@ def test_canary_byte(self):
                 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):
diff --git a/Modules/_testcapi/bytes.c b/Modules/_testcapi/bytes.c
index 83d1dab6c749f94..b2c7a9ca464e5eb 100644
--- a/Modules/_testcapi/bytes.c
+++ b/Modules/_testcapi/bytes.c
@@ -465,24 +465,19 @@ byteswriter_test_canary_byte(PyObject *Py_UNUSED(module), PyObject *args)
         return NULL;
     }
 
-    PyBytesWriter *writer = PyBytesWriter_Create(0);
+    PyBytesWriter *writer = PyBytesWriter_Create(len);
     if (writer == NULL) {
-        goto error;
-    }
-    if (PyBytesWriter_Grow(writer, len) < 0) {
-        goto error;
+        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);
-
-error:
-    PyBytesWriter_Discard(writer);
-    return NULL;
 }
 
 

From 7994e4ef758f91e9d213c1cacd4ed240055913d8 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Sat, 12 Sep 2026 19:32:36 +0200
Subject: [PATCH 4/9] Rewrite canary byte test in Python

* Add test_get_data_canary()
* Update test_get_data()
---
 Lib/test/test_capi/test_bytes.py | 71 +++++++++++++++++++++-----------
 Modules/_testcapi/bytes.c        | 57 +++++++++----------------
 2 files changed, 68 insertions(+), 60 deletions(-)

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index 7cb398a546f4fd2..cdcd4adfe08cb17 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -324,6 +324,7 @@ class BaseWriterTest:
     assert SMALL_BUFFER < _testcapi.PyBytesWriter_small_buffer
     LARGE_BUFFER = _testcapi.PyBytesWriter_small_buffer + 17  # bytes
     NEW_BYTE = b'\xff'
+    CANARY_BYTE = b'\xdd'
 
     def create_writer(self, alloc=0, string=b''):
         raise NotImplementedError
@@ -346,6 +347,7 @@ def test_get_data(self):
         # Test PyBytesWriter_GetData()
         writer = self.create_writer(6)
         NEW_BYTE = self.NEW_BYTE
+        CANARY_BYTE = self.CANARY_BYTE
         self.assertEqual(writer.get_data(), NEW_BYTE * 6)
         writer.write(0, b'abc')
         self.assertEqual(writer.get_data(), b'abc' + NEW_BYTE * 3)
@@ -359,7 +361,7 @@ def test_get_data(self):
         writer.write(0, b's' * small)
         self.assertEqual(writer.get_data(), b's' * small)
         writer.resize(large)
-        self.assertEqual(writer.get_data(), b's' * small + NEW_BYTE * (large - small))
+        self.assertEqual(writer.get_data(), b's' * small + CANARY_BYTE + NEW_BYTE * (large - small - 1))
         writer.write(small, b'L' * (large - small))
         self.assertEqual(writer.get_data(), b's' * small + b'L' * (large - small))
 
@@ -515,6 +517,51 @@ def test_format_i(self):
         writer.format_i(b'y=%i', 456)
         self.assertEqual(writer.finish(), b'x=123, y=456')
 
+    @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
+        use_bytearray = (self.RESULT_TYPE == bytearray)
+
+        # Test small buffer and large buffer
+        for size in (0, self.SMALL_BUFFER, self.LARGE_BUFFER):
+            with self.subTest(size=size):
+                code = textwrap.dedent(f"""
+                    from test.support import SuppressCrashReport
+                    import _testcapi
+                    size = {size}
+                    # Add an extra '#' byte to trigger a buffer overflow
+                    data = b'x' * size + b'#'
+                    use_bytearray = {use_bytearray}
+                    writer = _testcapi.PyBytesWriter(size, use_bytearray)
+                    with SuppressCrashReport():
+                        writer.write(0, data, check=False)
+                        writer.finish()
+                """)
+                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)
+
+    @unittest.skipUnless(support.Py_DEBUG, 'need Py_DEBUG')
+    def test_get_data_canary(self):
+        # Test PyBytesWriter_GetData()
+        NEW_BYTE = self.NEW_BYTE
+        CANARY_BYTE = self.CANARY_BYTE
+        canary_byte_size = len(CANARY_BYTE)
+
+        def get_data_canary():
+            size = writer.get_size() + canary_byte_size
+            return writer.get_data(size)
+
+        writer = self.create_writer(6)
+        self.assertEqual(get_data_canary(), NEW_BYTE * 6 + CANARY_BYTE)
+        writer.write(0, b'abc')
+        self.assertEqual(get_data_canary(), b'abc' + NEW_BYTE * 3 + CANARY_BYTE)
+        writer.write(3, b'123')
+        self.assertEqual(get_data_canary(), b'abc123' + CANARY_BYTE)
+
 
 class BytesWriterTest(BaseWriterTest, unittest.TestCase):
     RESULT_TYPE = bytes
@@ -567,28 +614,6 @@ 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
diff --git a/Modules/_testcapi/bytes.c b/Modules/_testcapi/bytes.c
index 5182d58a401e256..b4468dff0d0ba0d 100644
--- a/Modules/_testcapi/bytes.c
+++ b/Modules/_testcapi/bytes.c
@@ -135,22 +135,29 @@ writer_check(WriterObject *self)
 
 
 static PyObject*
-writer_write(PyObject *self_raw, PyObject *args)
+writer_write(PyObject *self_raw, PyObject *args, PyObject *kwargs)
 {
     WriterObject *self = (WriterObject *)self_raw;
     if (writer_check(self) < 0) {
         return NULL;
     }
 
+    static char *kwlist[] = {"pos", "str", "check", NULL};
     Py_ssize_t pos, size;
     char *str;
-    if (!PyArg_ParseTuple(args, "ny#", &pos, &str, &size)) {
+    int check = 1;
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs,
+                                     "ny#|i", kwlist,
+                                     &pos, &str, &size, &check)) {
         return NULL;
     }
 
-    if (pos < 0 || (pos + size) > PyBytesWriter_GetSize(self->writer)) {
-        PyErr_SetString(PyExc_ValueError, "invalid position or size");
-        return NULL;
+    // Use check=0 to trigger a buffer overflow for example
+    if (check) {
+        if (pos < 0 || (pos + size) > PyBytesWriter_GetSize(self->writer)) {
+            PyErr_SetString(PyExc_ValueError, "invalid position or size");
+            return NULL;
+        }
     }
 
     char *data = PyBytesWriter_GetData(self->writer);
@@ -245,15 +252,19 @@ writer_grow(PyObject *self_raw, PyObject *args)
 
 
 static PyObject*
-writer_get_data(PyObject *self_raw, PyObject *Py_UNUSED(args))
+writer_get_data(PyObject *self_raw, PyObject *args)
 {
     WriterObject *self = (WriterObject *)self_raw;
     if (writer_check(self) < 0) {
         return NULL;
     }
 
-    const char *data = PyBytesWriter_GetData(self->writer);
     Py_ssize_t size = PyBytesWriter_GetSize(self->writer);
+    if (!PyArg_ParseTuple(args, "|n", &size)) {
+        return NULL;
+    }
+
+    const char *data = PyBytesWriter_GetData(self->writer);
     return PyBytes_FromStringAndSize(data, size);
 }
 
@@ -305,12 +316,12 @@ writer_finish_with_size(PyObject *self_raw, PyObject *args)
 
 
 static PyMethodDef writer_methods[] = {
-    {"write", _PyCFunction_CAST(writer_write), METH_VARARGS},
+    {"write", _PyCFunction_CAST(writer_write), METH_VARARGS | METH_KEYWORDS},
     {"write_bytes", _PyCFunction_CAST(writer_write_bytes), METH_VARARGS},
     {"format_i", _PyCFunction_CAST(writer_format_i), METH_VARARGS},
     {"resize", _PyCFunction_CAST(writer_resize), METH_VARARGS},
     {"grow", _PyCFunction_CAST(writer_grow), METH_VARARGS},
-    {"get_data", _PyCFunction_CAST(writer_get_data), METH_NOARGS},
+    {"get_data", _PyCFunction_CAST(writer_get_data), METH_VARARGS},
     {"get_size", _PyCFunction_CAST(writer_get_size), METH_NOARGS},
     {"finish", _PyCFunction_CAST(writer_finish), METH_NOARGS},
     {"finish_with_size", _PyCFunction_CAST(writer_finish_with_size), METH_VARARGS},
@@ -502,33 +513,6 @@ 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},
@@ -536,7 +520,6 @@ static PyMethodDef test_methods[] = {
     {"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},
 };
 

From 19fdcb40ea031556e80d283be95b3a1d0c4af7b1 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Sat, 12 Sep 2026 19:43:15 +0200
Subject: [PATCH 5/9] Change fatal error message

---
 Objects/bytesobject.c | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/Objects/bytesobject.c b/Objects/bytesobject.c
index 274e89d2b411f77..9c72b5886a83d04 100644
--- a/Objects/bytesobject.c
+++ b/Objects/bytesobject.c
@@ -3624,9 +3624,8 @@ byteswriter_check_canary_byte(PyBytesWriter *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)",
+                             "Buffer overflow detected in PyBytesWriter %p "
+                             "at position %zd",
                              writer, writer->size);
     }
 }

From dc79819cc59108c7c8198253cad92d2a8e4eefd2 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Sat, 12 Sep 2026 22:16:40 +0200
Subject: [PATCH 6/9] gh-156939: Do no modify writer->size directly in UCS1
 encoder

unicode_encode_ucs1() now calls PyBytesWriter_Grow() to update the
PyBytesWriter size.
---
 Objects/unicodeobject.c | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/Objects/unicodeobject.c b/Objects/unicodeobject.c
index e86291347c75be0..86b9baadd0d8aa9 100644
--- a/Objects/unicodeobject.c
+++ b/Objects/unicodeobject.c
@@ -796,6 +796,8 @@ backslashreplace(PyBytesWriter *writer, char *str,
         }
         size += incr;
     }
+    /* subtract preallocated bytes */
+    size -= (collend - collstart);
 
     str = PyBytesWriter_GrowAndUpdatePointer(writer, size, str);
     if (str == NULL) {
@@ -871,6 +873,8 @@ xmlcharrefreplace(PyBytesWriter *writer, char *str,
         }
         size += incr;
     }
+    /* subtract preallocated bytes */
+    size -= (collend - collstart);
 
     str = PyBytesWriter_GrowAndUpdatePointer(writer, size, str);
     if (str == NULL) {
@@ -7262,8 +7266,6 @@ unicode_encode_ucs1(PyObject *unicode,
                 break;
 
             case _Py_ERROR_BACKSLASHREPLACE:
-                /* subtract preallocated bytes */
-                writer->size -= (collend - collstart);
                 str = backslashreplace(writer, str,
                                        unicode, collstart, collend);
                 if (str == NULL)
@@ -7272,8 +7274,6 @@ unicode_encode_ucs1(PyObject *unicode,
                 break;
 
             case _Py_ERROR_XMLCHARREFREPLACE:
-                /* subtract preallocated bytes */
-                writer->size -= (collend - collstart);
                 str = xmlcharrefreplace(writer, str,
                                         unicode, collstart, collend);
                 if (str == NULL)
@@ -7314,10 +7314,13 @@ unicode_encode_ucs1(PyObject *unicode,
                     }
                 }
                 else {
-                    /* subtract preallocated bytes */
-                    writer->size -= newpos - collstart;
                     /* Only overallocate the buffer if it's not the last write */
                     writer->overallocate = (newpos < size);
+
+                    /* subtract preallocated bytes */
+                    if (PyBytesWriter_Grow(writer, -(newpos - collstart)) < 0) {
+                        goto onError;
+                    }
                 }
 
                 const char *rep_str;

From e14b81ceeae25e4ada8a61bd82656dbdaf7c9642 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Sat, 12 Sep 2026 22:20:34 +0200
Subject: [PATCH 7/9] Update get_data() tests for canary byte

---
 Lib/test/test_capi/test_bytes.py | 43 ++++++++++++++++++++++++--------
 1 file changed, 33 insertions(+), 10 deletions(-)

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index 0a1a876417616b6..764ea8f81630a25 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -318,6 +318,11 @@ def test_join(self):
             bytes_join(b'', NULL)
 
 
+def get_data_canary(writer):
+    size = writer.get_size() + 1
+    return writer.get_data(size)
+
+
 class BaseWriterTest:
     RESULT_TYPE = NotImplementedError
     SMALL_BUFFER = 11  # bytes
@@ -426,6 +431,25 @@ def test_resize(self):
         writer.resize(len(b'number=123'))  # noop
         self.assertEqual(writer.finish(), b'number=123')
 
+        CANARY_BYTE = self.CANARY_BYTE
+        for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
+            with self.subTest(size=size):
+                # Truncate the last byte
+                data = b'x' * size
+                writer = self.create_writer(size)
+                writer.write(0, data)
+                self.assertEqual(get_data_canary(writer), data + CANARY_BYTE)
+                writer.resize(size - 1)
+                self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE)
+                self.assertEqual(writer.finish(),  data[:-1])
+
+                # Make the buffer empty
+                writer = self.create_writer(size)
+                writer.write(0, data)
+                writer.resize(0)
+                self.assertEqual(writer.get_data(), b'')
+                self.assertEqual(writer.finish(),  b'')
+
         # Switch from small buffer to large buffer
         writer = self.create_writer()
         small, large = self.SMALL_BUFFER, self.LARGE_BUFFER
@@ -465,15 +489,16 @@ def test_grow(self):
         writer.grow(0)  # noop
         self.assertEqual(writer.finish(), b'number=123')
 
+        CANARY_BYTE = self.CANARY_BYTE
         for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
             with self.subTest(size=size):
                 # Truncate the last byte
                 data = b'x' * size
                 writer = self.create_writer(size)
                 writer.write(0, data)
-                self.assertEqual(writer.get_data(), data)
+                self.assertEqual(get_data_canary(writer), data + CANARY_BYTE)
                 writer.grow(-1)
-                self.assertEqual(writer.get_data(), data[:-1])
+                self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE)
                 self.assertEqual(writer.finish(),  data[:-1])
 
                 # Make the buffer empty
@@ -567,18 +592,16 @@ def test_get_data_canary(self):
         # Test PyBytesWriter_GetData()
         NEW_BYTE = self.NEW_BYTE
         CANARY_BYTE = self.CANARY_BYTE
-        canary_byte_size = len(CANARY_BYTE)
-
-        def get_data_canary():
-            size = writer.get_size() + canary_byte_size
-            return writer.get_data(size)
 
         writer = self.create_writer(6)
-        self.assertEqual(get_data_canary(), NEW_BYTE * 6 + CANARY_BYTE)
+        self.assertEqual(get_data_canary(writer),
+                         NEW_BYTE * 6 + CANARY_BYTE)
         writer.write(0, b'abc')
-        self.assertEqual(get_data_canary(), b'abc' + NEW_BYTE * 3 + CANARY_BYTE)
+        self.assertEqual(get_data_canary(writer),
+                         b'abc' + NEW_BYTE * 3 + CANARY_BYTE)
         writer.write(3, b'123')
-        self.assertEqual(get_data_canary(), b'abc123' + CANARY_BYTE)
+        self.assertEqual(get_data_canary(writer),
+                         b'abc123' + CANARY_BYTE)
 
 
 class BytesWriterTest(BaseWriterTest, unittest.TestCase):

From e0299319be85c15af72dee645a20614300b5853e Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Sun, 13 Sep 2026 01:48:08 +0200
Subject: [PATCH 8/9] Skip canary byte tests on release build

---
 Lib/test/test_capi/test_bytes.py | 83 +++++++++++++++++---------------
 1 file changed, 44 insertions(+), 39 deletions(-)

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index 764ea8f81630a25..21643fd8f756e83 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -431,25 +431,6 @@ def test_resize(self):
         writer.resize(len(b'number=123'))  # noop
         self.assertEqual(writer.finish(), b'number=123')
 
-        CANARY_BYTE = self.CANARY_BYTE
-        for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
-            with self.subTest(size=size):
-                # Truncate the last byte
-                data = b'x' * size
-                writer = self.create_writer(size)
-                writer.write(0, data)
-                self.assertEqual(get_data_canary(writer), data + CANARY_BYTE)
-                writer.resize(size - 1)
-                self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE)
-                self.assertEqual(writer.finish(),  data[:-1])
-
-                # Make the buffer empty
-                writer = self.create_writer(size)
-                writer.write(0, data)
-                writer.resize(0)
-                self.assertEqual(writer.get_data(), b'')
-                self.assertEqual(writer.finish(),  b'')
-
         # Switch from small buffer to large buffer
         writer = self.create_writer()
         small, large = self.SMALL_BUFFER, self.LARGE_BUFFER
@@ -471,6 +452,27 @@ def test_resize(self):
                     writer.resize(_testcapi.PY_SSIZE_T_MAX)
                 self.assertEqual(writer.finish(), b'x' * size)
 
+    @unittest.skipUnless(support.Py_DEBUG, 'need debug build')
+    def test_resize_canary(self):
+        CANARY_BYTE = self.CANARY_BYTE
+        for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
+            with self.subTest(size=size):
+                # Truncate the last byte
+                data = b'x' * size
+                writer = self.create_writer(size)
+                writer.write(0, data)
+                self.assertEqual(get_data_canary(writer), data + CANARY_BYTE)
+                writer.resize(size - 1)
+                self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE)
+                self.assertEqual(writer.finish(),  data[:-1])
+
+                # Make the buffer empty
+                writer = self.create_writer(size)
+                writer.write(0, data)
+                writer.resize(0)
+                self.assertEqual(writer.get_data(), b'')
+                self.assertEqual(writer.finish(),  b'')
+
     def test_grow(self):
         # Test PyBytesWriter_Grow()
         writer = self.create_writer(0)
@@ -489,25 +491,6 @@ def test_grow(self):
         writer.grow(0)  # noop
         self.assertEqual(writer.finish(), b'number=123')
 
-        CANARY_BYTE = self.CANARY_BYTE
-        for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
-            with self.subTest(size=size):
-                # Truncate the last byte
-                data = b'x' * size
-                writer = self.create_writer(size)
-                writer.write(0, data)
-                self.assertEqual(get_data_canary(writer), data + CANARY_BYTE)
-                writer.grow(-1)
-                self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE)
-                self.assertEqual(writer.finish(),  data[:-1])
-
-                # Make the buffer empty
-                writer = self.create_writer(size)
-                writer.write(0, data)
-                writer.grow(-size)
-                self.assertEqual(writer.get_data(), b'')
-                self.assertEqual(writer.finish(),  b'')
-
         # Switch from small buffer to large buffer
         writer = self.create_writer()
         small, large = self.SMALL_BUFFER, self.LARGE_BUFFER
@@ -529,6 +512,28 @@ def test_grow(self):
                     writer.grow(_testcapi.PY_SSIZE_T_MAX)
                 self.assertEqual(writer.finish(), b'x' * size)
 
+    @unittest.skipUnless(support.Py_DEBUG, 'need debug build')
+    def test_grow_canary(self):
+        CANARY_BYTE = self.CANARY_BYTE
+        for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
+            with self.subTest(size=size):
+                # Truncate the last byte
+                data = b'x' * size
+                writer = self.create_writer(size)
+                writer.write(0, data)
+                self.assertEqual(get_data_canary(writer), data + CANARY_BYTE)
+                writer.grow(-1)
+                self.assertEqual(get_data_canary(writer), data[:-1] + CANARY_BYTE)
+                self.assertEqual(writer.finish(),  data[:-1])
+
+                # Make the buffer empty
+                writer = self.create_writer(size)
+                writer.write(0, data)
+                writer.grow(-size)
+                self.assertEqual(writer.get_data(), b'')
+                self.assertEqual(writer.finish(),  b'')
+
+
     @support.nomemtest
     def test_resize_error(self):
         # Test PyBytesWriter_Resize() error
@@ -587,7 +592,7 @@ def test_canary_byte(self):
                 self.assertIn(f'at position {size}'.encode(),
                               proc.err)
 
-    @unittest.skipUnless(support.Py_DEBUG, 'need Py_DEBUG')
+    @unittest.skipUnless(support.Py_DEBUG, 'need debug build')
     def test_get_data_canary(self):
         # Test PyBytesWriter_GetData()
         NEW_BYTE = self.NEW_BYTE

From 276c315b30a8eca713d2e47a6cca9ed741b710e6 Mon Sep 17 00:00:00 2001
From: Victor Stinner <vstinner@python.org>
Date: Sun, 13 Sep 2026 01:53:33 +0200
Subject: [PATCH 9/9] Add test_grow_error()

PyBytesWriter_Grow() no longer calls byteswriter_resize() if grow is
smaller than 0.
---
 Lib/test/test_capi/test_bytes.py | 32 +++++++++++++++++++++++++-------
 Objects/bytesobject.c            | 25 ++++++++++++++-----------
 2 files changed, 39 insertions(+), 18 deletions(-)

diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
index 21643fd8f756e83..a500f2c702db0fb 100644
--- a/Lib/test/test_capi/test_bytes.py
+++ b/Lib/test/test_capi/test_bytes.py
@@ -473,6 +473,26 @@ def test_resize_canary(self):
                 self.assertEqual(writer.get_data(), b'')
                 self.assertEqual(writer.finish(),  b'')
 
+    @support.nomemtest
+    def test_resize_error(self):
+        # Test PyBytesWriter_Resize() error
+        init = b'x' * self.LARGE_BUFFER
+        writer = self.create_writer(len(init))
+        writer.write(0, init)
+        size = len(init) + 100
+        try:
+            with self.assertRaises(MemoryError):
+                _testcapi.set_nomemory(0)
+                writer.resize(size)
+        finally:
+            _testcapi.remove_mem_hooks()
+        suffix = b'still working'
+        writer.write_bytes(suffix, -1)
+        self.assertEqual(writer.finish(), init + suffix)
+
+        # Note: PyBytesWriter_Resize() leaves the buffer unchanged (no resize)
+        # if the new size is smaller than the allocated size
+
     def test_grow(self):
         # Test PyBytesWriter_Grow()
         writer = self.create_writer(0)
@@ -533,26 +553,24 @@ def test_grow_canary(self):
                 self.assertEqual(writer.get_data(), b'')
                 self.assertEqual(writer.finish(),  b'')
 
-
     @support.nomemtest
-    def test_resize_error(self):
-        # Test PyBytesWriter_Resize() error
+    def test_grow_error(self):
+        # Test PyBytesWriter_Grow() error
         init = b'x' * self.LARGE_BUFFER
         writer = self.create_writer(len(init))
         writer.write(0, init)
-        size = len(init) + 100
         try:
             with self.assertRaises(MemoryError):
                 _testcapi.set_nomemory(0)
-                writer.resize(size)
+                writer.grow(100)
         finally:
             _testcapi.remove_mem_hooks()
         suffix = b'still working'
         writer.write_bytes(suffix, -1)
         self.assertEqual(writer.finish(), init + suffix)
 
-        # Note: PyBytesWriter_Resize() leaves the buffer unchanged (no resize)
-        # if the new size is smaller than the allocated size
+        # Note: PyBytesWriter_Grow() leaves the buffer unchanged (no resize)
+        # if grow is negative.
 
     def test_format_i(self):
         # Test PyBytesWriter_Format()
diff --git a/Objects/bytesobject.c b/Objects/bytesobject.c
index beaf3b45b6db302..10be2791a38e538 100644
--- a/Objects/bytesobject.c
+++ b/Objects/bytesobject.c
@@ -3883,21 +3883,21 @@ PyBytesWriter_GetSize(PyBytesWriter *writer)
 
 
 int
-PyBytesWriter_Resize(PyBytesWriter *writer, Py_ssize_t size)
+PyBytesWriter_Resize(PyBytesWriter *writer, Py_ssize_t new_size)
 {
-    if (size < 0) {
+    if (new_size < 0) {
         PyErr_SetString(PyExc_ValueError, "size must be >= 0");
         return -1;
     }
-    if (writer->size < size) {
-        if (byteswriter_resize(writer, size, 1) < 0) {
+    if (writer->size < new_size) {
+        if (byteswriter_resize(writer, new_size, 1) < 0) {
             return -1;
         }
     }
     else {
         // The buffer is already large enough. Never shrink the buffer.
     }
-    writer->size = size;
+    writer->size = new_size;
 #ifdef Py_DEBUG
     byteswriter_write_canary_byte(writer);
 #endif
@@ -3925,24 +3925,27 @@ PyBytesWriter_Grow(PyBytesWriter *writer, Py_ssize_t grow)
         return 0;
     }
 
-    if (grow >= 0) {
+    if (grow > 0) {
         if (grow > PY_SSIZE_T_MAX - writer->size) {
             PyErr_NoMemory();
             return -1;
         }
+        Py_ssize_t new_size = writer->size + grow;
+
+        if (byteswriter_resize(writer, new_size, 1) < 0) {
+            return -1;
+        }
+        writer->size = new_size;
     }
     else {
         if (writer->size + grow < 0) {
             PyErr_SetString(PyExc_ValueError, "invalid size");
             return -1;
         }
+        // The buffer is already large enough. Never shrink the buffer.
+        writer->size = writer->size + grow;
     }
-    Py_ssize_t size = writer->size + grow;
 
-    if (byteswriter_resize(writer, size, 1) < 0) {
-        return -1;
-    }
-    writer->size = size;
 #ifdef Py_DEBUG
     byteswriter_write_canary_byte(writer);
 #endif