[pull] main from python:main

Doc/c-api/bytes.rst

@@ -305,8 +305,8 @@ object.
 
    A bytes writer object.
 
-   The API is **not thread safe**. A :c:type:`PyBytesWriter` object must only
-   be used by a single thread, it must not be shared between threads.
+   The API is **not thread safe**. To share a writer with multiple threads, a
+   critical section or a lock is needed.
 
    The instance must be destroyed by :c:func:`PyBytesWriter_Finish` on
    success, or :c:func:`PyBytesWriter_Discard` on error.

Doc/c-api/unicode.rst

@@ -1797,6 +1797,9 @@ object.
    The instance must be destroyed by :c:func:`PyUnicodeWriter_Finish` on
    success, or :c:func:`PyUnicodeWriter_Discard` on error.
 
+   The API is **not thread safe**. To share a writer with multiple threads, a
+   critical section or a lock is needed.
+
 .. c:function:: PyUnicodeWriter* PyUnicodeWriter_Create(Py_ssize_t length)
 
    Create a Unicode writer instance.

Include/internal/pycore_unicodeobject.h

@@ -16,7 +16,9 @@ extern "C" {
 #define _Py_MAX_UNICODE 0x10ffff
 
 
-extern int _PyUnicode_IsModifiable(PyObject *unicode);
+// Export for '_multibytecodec' shared extension. _PyUnicodeWriter_CanWrite()
+// calls this function when assertions are enabled.
+PyAPI_FUNC(int) _PyUnicode_IsModifiable(PyObject *unicode);
 extern void _PyUnicodeWriter_InitWithBuffer(
     _PyUnicodeWriter *writer,
     PyObject *buffer);
@@ -105,12 +107,32 @@ _PyUnicode_EnsureUnicode(PyObject *obj)
     return 0;
 }
 
+#ifndef NDEBUG
+static inline int
+_PyUnicodeWriter_CanWrite(_PyUnicodeWriter *writer)
+{
+    // Code adapted from _PyUnicode_IsModifiable()
+    assert(!writer->readonly);
+    PyObject *buffer = writer->buffer;
+    assert(buffer != NULL);
+    // Do not use _PyObject_IsUniquelyReferenced(): the caller can have its own
+    // lock to prevent a writer from being used by two threads at the same
+    // time.
+    assert(Py_REFCNT(buffer) == 1);
+    assert(PyUnstable_Unicode_GET_CACHED_HASH(buffer) == -1);
+    assert(!PyUnicode_CHECK_INTERNED(buffer));
+    assert(!_Py_IsImmortal(buffer));
+    return 1;
+}
+#endif
+
 static inline int
 _PyUnicodeWriter_WriteCharInline(_PyUnicodeWriter *writer, Py_UCS4 ch)
 {
     assert(ch <= _Py_MAX_UNICODE);
     if (_PyUnicodeWriter_Prepare(writer, 1, ch) < 0)
         return -1;
+    assert(_PyUnicodeWriter_CanWrite(writer));
     PyUnicode_WRITE(writer->kind, writer->data, writer->pos, ch);
     writer->pos++;
     return 0;

Lib/test/test_capi/test_bytes.py

@@ -1,8 +1,10 @@
 import sys
 import textwrap
+import threading
 import unittest
 from test import support
 from test.support import import_helper
+from test.support import threading_helper
 from test.support.script_helper import assert_python_failure
 
 _testlimitedcapi = import_helper.import_module('_testlimitedcapi')
@@ -461,6 +463,14 @@ def test_resize(self):
         self.assertEqual(writer.finish(),
                          b's' * small + b'L' * (large - small))
 
+        # Make sure that it's possible to write after a resize to zero
+        # when a bytes/bytearray object is allocated.
+        writer = self.create_writer()
+        writer.resize(self.LARGE_BUFFER)
+        writer.resize(0)
+        writer.write_bytes(b'abc', 3)
+        self.assertEqual(writer.finish(), b'abc')
+
         # invalid size
         for size in (self.SMALL_BUFFER, self.LARGE_BUFFER):
             with self.subTest(size=size):
@@ -665,6 +675,36 @@ def test_get_data_canary(self):
         self.assertEqual(get_data_canary(writer),
                          b'abc123' + CANARY_BYTE)
 
+    @threading_helper.requires_working_threading()
+    def test_thread(self):
+        # PyBytesWriter can be used by multiple threads: it's up to the caller
+        # to implement a lock to prevent concurrent accesses.
+        writer = self.create_writer(0)
+        size = None
+        data = None
+
+        def thread_func(writer, LARGE_BUFFER):
+            nonlocal size, data
+
+            # create a bytes object for the buffer
+            writer.write_bytes(b'x' * LARGE_BUFFER, LARGE_BUFFER)
+
+            # so we can check a write with a bytes object
+            writer.write_bytes(b'yz', 2)
+            writer.format_i(b'i=%i', 5)
+            writer.resize(10)
+            data = writer.get_data()
+            size = writer.get_size()
+
+        thread = threading.Thread(target=thread_func,
+                                  args=(writer, self.LARGE_BUFFER))
+        thread.start()
+        threading_helper.join_thread(thread)
+
+        self.assertEqual(size, 10)
+        self.assertEqual(data, b'x' * 10)
+        self.assertEqual(writer.finish(), b'x' * 10)
+
 
 class BytesWriterTest(BaseWriterTest, unittest.TestCase):
     RESULT_TYPE = bytes

Lib/test/test_capi/test_unicode.py

@@ -1,7 +1,9 @@
-import unittest
 import sys
+import textwrap
+import unittest
 from test import support
 from test.support import threading_helper
+from test.support.script_helper import assert_python_failure
 
 try:
     import _testcapi
@@ -1981,6 +1983,33 @@ def test_substring_empty(self):
         writer.write_substring("abc", 1, 1)
         self.assertEqual(writer.finish(), '')
 
+    def test_singletons(self):
+        writer = self.create_writer(5)
+        self.assertIs(writer.finish(), '')
+
+        for ch in range(256):
+            with self.subTest(ch=ch):
+                ch = chr(ch)
+                writer = self.create_writer(0)
+                writer.write_substring(ch + 'xxx', 0, 1)
+                self.assertIs(writer.finish(), ch)
+
+    @unittest.skipUnless(support.Py_DEBUG, 'need debug build (Py_DEBUG)')
+    def test_detect_overflow(self):
+        # Test detection of buffer overflow
+        code = textwrap.dedent('''
+            from test.support import SuppressCrashReport
+            import _testinternalcapi
+
+            SuppressCrashReport().__enter__()
+            _testinternalcapi.unicodewriter_overflow()
+        ''')
+        proc = assert_python_failure('-c', code)
+        self.assertIn(b'Buffer overflow detected in PyUnicodeWriter', proc.err)
+        # Do not test the position value since it depends on the overallocation
+        # strategy which depends on the operating system
+        self.assertIn(f'at position '.encode(), proc.err)
+
 
 @unittest.skipIf(ctypes is None, 'need ctypes')
 class PyUnicodeWriterFormatTest(unittest.TestCase):

Misc/NEWS.d/next/C_API/2026-09-18-00-29-49.gh-issue-157710.Q_AD8D.rst

@@ -0,0 +1,3 @@
+When Python is built in debug mode, :c:func:`PyUnicodeWriter_Finish` now
+checks if the trailing null byte has been overridden to detect buffer
+overflow. Patch by Victor Stinner.

Modules/_testinternalcapi.c

@@ -3206,6 +3206,28 @@ test_thread_state_ensure_from_view_interp_switch(PyObject *self, PyObject *unuse
     Py_RETURN_NONE;
 }
 
+static PyObject *
+unicodewriter_overflow(PyObject *self, PyObject *unused)
+{
+    PyUnicodeWriter *writer = PyUnicodeWriter_Create(0);
+    if (writer == NULL) {
+        return NULL;
+    }
+    if (PyUnicodeWriter_WriteASCII(writer, "hello", -1) < 0) {
+        PyUnicodeWriter_Discard(writer);
+        return NULL;
+    }
+
+    _PyUnicodeWriter *impl = (_PyUnicodeWriter*)writer;
+    PyObject *buffer = impl->buffer;
+    Py_ssize_t index = PyUnicode_GET_LENGTH(buffer);
+    PyUnicode_WRITE(impl->kind, impl->data, index, '#');  // overflow!
+
+    // Spoiler: the function doesn't return if an overflow is detected
+    // in debug mode
+    return PyUnicodeWriter_Finish(writer);
+}
+
 /* Self interrupting context manager */
 
 typedef struct {
@@ -3393,6 +3415,7 @@ static PyMethodDef module_functions[] = {
     {"test_interp_guard_countdown", test_interp_guard_countdown, METH_NOARGS},
     {"test_interp_view_countdown", test_interp_view_countdown, METH_NOARGS},
     {"test_thread_state_ensure_from_view_interp_switch", test_thread_state_ensure_from_view_interp_switch, METH_NOARGS},
+    {"unicodewriter_overflow", unicodewriter_overflow, METH_NOARGS},
     {NULL, NULL} /* sentinel */
 };