Merge branch 'main' into writer_create

Author: vstinner

Lib/test/test_bytes.py

@@ -520,6 +520,15 @@ def test_fromhex(self):
                 self.type2test.fromhex(data)
             self.assertIn('at position %s' % pos, str(cm.exception))
 
+        # gh-158583: Check for out of bounds reads (uninitialized bytes).
+        # Create an array from a list to not overallocate.
+        a = array.array('B', list(b'1234  '))  # Py_ISSPACE() loop
+        self.assertEqual(self.type2test.fromhex(a), b'\x12\x34')
+
+        a = array.array('B', list(b'12345'))  # Missing second digit
+        with self.assertRaises(ValueError):
+            self.type2test.fromhex(a)
+
     def test_hex(self):
         self.assertRaises(TypeError, self.type2test.hex)
         self.assertRaises(TypeError, self.type2test.hex, 1)

Lib/test/test_capi/test_bytes.py

@@ -244,39 +244,60 @@ def test_resize(self):
         """Test _PyBytes_Resize()"""
         _resize = _testcapi.bytes_resize
 
-        def resize(obj, size, new):
-            result = _resize(obj, size, new)
-            if 1 <= len(result):
-                if new or size != len(obj):
-                    # gh-156995: Make sure that the result is a fresh object.
-                    # Previously, _PyBytes_Resize(&obj, 1) returned a singleton
-                    # if _PyObject_IsUniquelyReferenced() is false.
-                    self.assertEqual(sys.getrefcount(result), 1)
-                    self.assertFalse(sys._is_immortal(result))
-            else:
+        def assert_is_fresh_copy(result, refcnt, is_new_obj):
+            self.assertEqual(refcnt, 1)
+            self.assertTrue(is_new_obj)
+            self.assertFalse(sys._is_immortal(result))
+
+        def resize(obj, size, new, compute_hash=False):
+            old_size = len(obj)
+            result, refcnt, is_new_obj = _resize(obj, size, new,
+                                                 compute_hash=compute_hash)
+
+            if size == old_size:
+                # Return the same object unchanged
+                self.assertFalse(is_new_obj)
+            elif old_size == 0:
+                assert_is_fresh_copy(result, refcnt, is_new_obj)
+            elif size == 0:
                 # check that the result is the empty bytes string singleton
+                self.assertEqual(result, b'')
                 self.assertTrue(sys._is_immortal(result))
+                self.assertTrue(is_new_obj)
+            elif (not new) or compute_hash:
+                # gh-156995: Make sure that the result is a fresh object.
+                # Previously, _PyBytes_Resize(&obj, 1) returned a singleton
+                # if _PyObject_IsUniquelyReferenced() is false.
+                assert_is_fresh_copy(result, refcnt, is_new_obj)
+            else:
+                # An in-place resize can return the same memory address, or
+                # not. 'is_new_obj' cannot be tested.
+                self.assertEqual(refcnt, 1)
+                self.assertFalse(sys._is_immortal(result))
+
             return result
 
         for new in True, False:
-            with self.subTest(new=new):
-                self.assertEqual(resize(b'abc', 0, new), b'')
-                self.assertEqual(resize(b'abc', 1, new), b'a')
-                self.assertEqual(resize(b'abc', 2, new), b'ab')
-                self.assertEqual(resize(b'abc', 3, new), b'abc')
-                b = resize(b'abc', 4, new)
-                self.assertEqual(len(b), 4)
-                self.assertEqual(b[:3], b'abc')
-
-                self.assertEqual(resize(b'a', 0, new), b'')
-                self.assertEqual(resize(b'a', 1, new), b'a')
-                b = resize(b'a', 2, new)
-                self.assertEqual(len(b), 2)
-                self.assertEqual(b[:1], b'a')
-
-                self.assertEqual(resize(b'', 0, new), b'')
-                self.assertEqual(len(resize(b'', 1, new)), 1)
-                self.assertEqual(len(resize(b'', 2, new)), 2)
+            for compute_hash in True, False:
+                with self.subTest(new=new, compute_hash=compute_hash):
+                    self.assertEqual(resize(b'abc', 0, new, compute_hash), b'')
+                    self.assertEqual(resize(b'abc', 1, new, compute_hash), b'a')
+                    self.assertEqual(resize(b'abc', 2, new, compute_hash), b'ab')
+                    self.assertEqual(resize(b'abc', 3, new, compute_hash), b'abc')
+
+                    b = resize(b'abc', 4, new, compute_hash)
+                    self.assertEqual(len(b), 4)
+                    self.assertEqual(b[:3], b'abc')
+
+                    self.assertEqual(resize(b'a', 0, new, compute_hash), b'')
+                    self.assertEqual(resize(b'a', 1, new, compute_hash), b'a')
+                    b = resize(b'a', 2, new, compute_hash)
+                    self.assertEqual(len(b), 2)
+                    self.assertEqual(b[:1], b'a')
+
+                    self.assertEqual(resize(b'', 0, new, compute_hash), b'')
+                    self.assertEqual(len(resize(b'', 1, new, compute_hash)), 1)
+                    self.assertEqual(len(resize(b'', 2, new, compute_hash)), 2)
 
         self.assertRaises(SystemError, resize, b'abc', -1, False)
         self.assertRaises(SystemError, resize, bytearray(b'abc'), 3, False)

Lib/test/test_capi/test_unicode.py

@@ -170,28 +170,75 @@ def test_write_macro(self):
 
     def test_resize(self):
         """Test PyUnicode_Resize()"""
-        resize = _testlimitedcapi.unicode_resize
+        _resize = _testlimitedcapi.unicode_resize
+        resize_null = _testlimitedcapi.unicode_resize_null
+
+        def resize(s, length, new=True, compute_hash=False):
+            if s is not NULL and isinstance(s, str):
+                old_length = len(s)
+            else:
+                old_length = 0
+            result, int_result, refcnt, is_new_obj = _resize(s, length,
+                                                             new, compute_hash)
+            self.assertEqual(int_result, 0)
+
+            if length == old_length:
+                # Return the same object unchanged
+                self.assertFalse(is_new_obj)
+            elif length == 0:
+                # Get the empty Unicode string
+                self.assertEqual(result, '')
+                self.assertTrue(sys._is_immortal(result))
+                self.assertTrue(is_new_obj)
+            elif (not new) or compute_hash:
+                # Get a fresh copy
+                self.assertEqual(refcnt, 1)
+                self.assertTrue(is_new_obj)
+                self.assertFalse(sys._is_immortal(result))
+            else:
+                # In-size replace can return the same address, or not.
+                # So 'is_new_obj' cannot be tested.
+                self.assertFalse(sys._is_immortal(result))
+
+            return result
 
         strings = [
             # all strings have exactly 3 characters
             'abc', '\xa1\xa2\xa3', '\u4f60\u597d\u4e16',
             '\U0001f600\U0001f601\U0001f602'
         ]
-        for s in strings:
-            self.assertEqual(resize(s, 3), (s, 0))
-            self.assertEqual(resize(s, 2), (s[:2], 0))
-            self.assertEqual(resize(s, 4), (s + '\0', 0))
-            self.assertEqual(resize(s, 10), (s + '\0'*7, 0))
-            self.assertEqual(resize(s, 0), ('', 0))
-            self.assertRaises(MemoryError, resize, s, PY_SSIZE_T_MAX)
-            self.assertRaises(SystemError, resize, s, -1)
-            self.assertRaises(SystemError, resize, s, PY_SSIZE_T_MIN)
+        for new in (True, False):
+            for compute_hash in (True, False):
+                for s in strings:
+                    with self.subTest(new=new, compute_hash=compute_hash, s=s):
+                        self.assertEqual(resize(s, 3, new, compute_hash),
+                                         s)
+                        self.assertEqual(resize(s, 2, new, compute_hash),
+                                         s[:2])
+                        self.assertEqual(resize(s, 4, new, compute_hash),
+                                         s + '\0')
+                        self.assertEqual(resize(s, 10, new, compute_hash),
+                                         s + '\0'*7)
+                        self.assertEqual(resize(s, 0, new, compute_hash),
+                                         '')
+
+                        with self.assertRaises(MemoryError):
+                            resize(s, PY_SSIZE_T_MAX, new, compute_hash)
+                        with self.assertRaises(SystemError):
+                            resize(s, -1, new, compute_hash)
+                        with self.assertRaises(SystemError):
+                            resize(s, PY_SSIZE_T_MIN, new, compute_hash)
+
         self.assertRaises(SystemError, resize, b'abc', 0)
         self.assertRaises(SystemError, resize, [], 0)
         self.assertRaises(SystemError, resize, NULL, 0)
         # TODO: Test PyUnicode_Resize() with non-modifiable and legacy unicode
         # and with NULL as the address.
 
+        # Test PyUnicode_Resize(NULL, length)
+        self.assertRaises(SystemError, resize_null, 0)
+        self.assertRaises(SystemError, resize_null, 123)
+
     def test_append(self):
         """Test PyUnicode_Append()"""
         append = _testlimitedcapi.unicode_append

Misc/NEWS.d/next/Core_and_Builtins/2026-10-03-16-14-26.gh-issue-158583.7EUMvS.rst

@@ -0,0 +1,2 @@
+:meth:`bytes.fromhex` and :meth:`bytearray.fromhex`: Fix uninitialized
+memory read. Patch by Victor Stinner.

Modules/_io/bufferedio.c

@@ -1205,109 +1205,115 @@ _io__Buffered_readinto1_impl(buffered *self, Py_buffer *buffer)
 static PyObject *
 _buffered_readline(buffered *self, Py_ssize_t limit)
 {
-    PyObject *res = NULL;
-    PyObject *chunks = NULL;
-    Py_ssize_t n;
-    const char *start, *s, *end;
+    _Py_CRITICAL_SECTION_ASSERT_OBJECT_LOCKED(self);
 
     CHECK_CLOSED(self, "readline of closed file")
 
     /* First, try to find a line in the buffer. This can run unlocked because
        the calls to the C API are simple enough that they can't trigger
        any thread switch. */
-    n = Py_SAFE_DOWNCAST(READAHEAD(self), Py_off_t, Py_ssize_t);
-    if (limit >= 0 && n > limit)
+    Py_ssize_t n = Py_SAFE_DOWNCAST(READAHEAD(self), Py_off_t, Py_ssize_t);
+    if (limit >= 0 && n > limit) {
         n = limit;
-    start = self->buffer + self->pos;
-    s = memchr(start, '\n', n);
+    }
+    const char *start = self->buffer + self->pos;
+    const char *s = memchr(start, '\n', n);
     if (s != NULL) {
-        res = PyBytes_FromStringAndSize(start, s - start + 1);
-        if (res != NULL)
-            self->pos += s - start + 1;
-        goto end_unlocked;
+        n = s - start + 1;
+        PyObject *res = PyBytes_FromStringAndSize(start, n);
+        if (res == NULL) {
+            return NULL;
+        }
+        self->pos += n;
+        return res;
     }
+
     if (n == limit) {
-        res = PyBytes_FromStringAndSize(start, n);
-        if (res != NULL)
-            self->pos += n;
-        goto end_unlocked;
+        PyObject *res = PyBytes_FromStringAndSize(start, n);
+        if (res == NULL) {
+            return NULL;
+        }
+        self->pos += n;
+        return res;
     }
 
-    if (!ENTER_BUFFERED(self))
-        goto end_unlocked;
+    PyBytesWriter *writer = NULL;
+    int locked = 0;
+    if (!ENTER_BUFFERED(self)) {
+        goto error;
+    }
+    locked = 1;
 
     /* Now we try to get some more from the raw stream */
-    chunks = PyList_New(0);
-    if (chunks == NULL)
-        goto end;
+    writer = PyBytesWriter_Create(0);
+    if (writer == NULL) {
+        goto error;
+    }
+
     if (n > 0) {
-        res = PyBytes_FromStringAndSize(start, n);
-        if (res == NULL)
-            goto end;
-        if (PyList_Append(chunks, res) < 0) {
-            Py_CLEAR(res);
-            goto end;
+        if (PyBytesWriter_WriteBytes(writer, start, n) < 0) {
+            goto error;
         }
-        Py_CLEAR(res);
         self->pos += n;
-        if (limit >= 0)
+        if (limit >= 0) {
             limit -= n;
+        }
     }
     if (self->writable) {
-        PyObject *r = buffered_flush_and_rewind_unlocked(self);
-        if (r == NULL)
-            goto end;
-        Py_DECREF(r);
+        PyObject *res = buffered_flush_and_rewind_unlocked(self);
+        if (res == NULL) {
+            goto error;
+        }
+        Py_DECREF(res);
     }
 
     for (;;) {
         _bufferedreader_reset_buf(self);
         n = _bufferedreader_fill_buffer(self);
-        if (n == -1)
-            goto end;
-        if (n <= 0)
+        if (n == -1) {
+            goto error;
+        }
+        if (n <= 0) {
             break;
-        if (limit >= 0 && n > limit)
+        }
+        if (limit >= 0 && n > limit) {
             n = limit;
+        }
         start = self->buffer;
-        end = start + n;
+        const char *end = start + n;
         s = start;
         while (s < end) {
             if (*s++ == '\n') {
-                res = PyBytes_FromStringAndSize(start, s - start);
-                if (res == NULL)
-                    goto end;
+                if (PyBytesWriter_WriteBytes(writer, start, s - start) < 0) {
+                    goto error;
+                }
                 self->pos = s - start;
                 goto found;
             }
         }
-        res = PyBytes_FromStringAndSize(start, n);
-        if (res == NULL)
-            goto end;
+
+        if (PyBytesWriter_WriteBytes(writer, start, n) < 0) {
+            goto error;
+        }
         if (n == limit) {
             self->pos = n;
             break;
         }
-        if (PyList_Append(chunks, res) < 0) {
-            Py_CLEAR(res);
-            goto end;
-        }
-        Py_CLEAR(res);
-        if (limit >= 0)
+        if (limit >= 0) {
             limit -= n;
+        }
     }
-found:
-    if (res != NULL && PyList_Append(chunks, res) < 0) {
-        Py_CLEAR(res);
-        goto end;
-    }
-    Py_XSETREF(res, PyBytes_Join((PyObject *)&_Py_SINGLETON(bytes_empty), chunks));
 
-end:
+found:
     LEAVE_BUFFERED(self)
-end_unlocked:
-    Py_XDECREF(chunks);
-    return res;
+    return PyBytesWriter_Finish(writer);
+
+error:
+    PyBytesWriter_Discard(writer);
+    if (locked) {
+        LEAVE_BUFFERED(self)
+    }
+    return NULL;
 }
 
 /*[clinic input]