diff --git a/Lib/test/test_capi/test_bytes.py b/Lib/test/test_capi/test_bytes.py
@@ -1,7 +1,9 @@
 import sys
+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')
@@ -316,12 +318,18 @@ 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
     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
@@ -344,6 +352,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)
@@ -357,7 +366,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))
 
@@ -443,6 +452,47 @@ 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'')
+
+    @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)
@@ -461,24 +511,6 @@ def test_grow(self):
         writer.grow(0)  # noop
         self.assertEqual(writer.finish(), b'number=123')
 
-        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)
-                writer.grow(-1)
-                self.assertEqual(writer.get_data(), data[:-1])
-                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
@@ -500,25 +532,45 @@ 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
+    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()
@@ -531,6 +583,49 @@ 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 debug build')
+    def test_get_data_canary(self):
+        # Test PyBytesWriter_GetData()
+        NEW_BYTE = self.NEW_BYTE
+        CANARY_BYTE = self.CANARY_BYTE
+
+        writer = self.create_writer(6)
+        self.assertEqual(get_data_canary(writer),
+                         NEW_BYTE * 6 + CANARY_BYTE)
+        writer.write(0, b'abc')
+        self.assertEqual(get_data_canary(writer),
+                         b'abc' + NEW_BYTE * 3 + CANARY_BYTE)
+        writer.write(3, b'123')
+        self.assertEqual(get_data_canary(writer),
+                         b'abc123' + CANARY_BYTE)
+
 
 class BytesWriterTest(BaseWriterTest, unittest.TestCase):
     RESULT_TYPE = bytes

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
@@ -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
@@ -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);
@@ -168,7 +175,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;
@@ -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},

diff --git 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
     }
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		When Python is built in debug mode, :c:type:`PyBytesWriter` now detects
		buffer overflow. Patch by Victor Stinner.