[3.14] gh-159739: Validate newline replacement in TextIOWrapper.write (GH-160191)

A str subclass can override replace() and return a non-string while TextIOWrapper.write() translates newlines. The C implementation then treats the result as Unicode and can crash. Reject an invalid replacement with TypeError. Backport to 3.15 and 3.14; on 3.14, put the regression test in Lib/test/test_io.py.
(cherry picked from commit c6796a1)

Co-authored-by: Jelle Zijlstra <jelle@openai.com>

Author: jelle-openai

Lib/test/test_io.py

@@ -4099,6 +4099,41 @@ class CTextIOWrapperTest(TextIOWrapperTest):
     io = io
     shutdown_error = "LookupError: unknown encoding: ascii"
 
+    def test_write_newline_replace_returns_non_string(self):
+        class Text(str):
+            def replace(self, *args):
+                return None
+
+        with self.TextIOWrapper(self.BytesIO(), newline='\r\n') as stream:
+            with self.assertRaisesRegex(TypeError, 'replace.*str'):
+                stream.write(Text('a\n'))
+
+    def test_write_newline_replace_changes_stream(self):
+        def fail_init(stream, buffer):
+            with self.assertRaises(LookupError):
+                stream.__init__(buffer, encoding='nonexistent-encoding')
+
+        actions = (
+            (lambda stream, buffer: stream.close(), ValueError, 'closed'),
+            (lambda stream, buffer: stream.detach(), ValueError, 'detached'),
+            (lambda stream, buffer: stream.__init__(buffer),
+             self.UnsupportedOperation, 'not writable'),
+            (fail_init, ValueError, 'uninitialized'),
+        )
+        for action, error, message in actions:
+            with self.subTest(message=message):
+                with (self.BytesIO() as raw,
+                      self.BufferedReader(self.BytesIO()) as readonly):
+                    stream = self.TextIOWrapper(raw, newline='\r\n')
+
+                    class Text(str):
+                        def replace(self, *args):
+                            action(stream, readonly)
+                            return 'a\r\n'
+
+                    with self.assertRaisesRegex(error, message):
+                        stream.write(Text('a\n'))
+
     def test_reentrant_seek_during_tell(self):
         # gh-153539: reading short of _CHUNK_SIZE leaves residual bytes in the
         # snapshot, so tell() re-decodes and calls the decoder's getstate(); a

Misc/NEWS.d/next/Library/2026-10-21-21-26-10.gh-issue-159739.k8uuGe.rst

@@ -0,0 +1,3 @@
+Raise :exc:`TypeError` if a string subclass returns a non-string during newline
+translation in :class:`io.TextIOWrapper`. Recheck the stream state if the
+replacement callback closes, detaches or reinitializes the stream.

Modules/_io/textio.c

@@ -1722,6 +1722,19 @@ _textiowrapper_writeflush(textio *self)
     return 0;
 }
 
+/* Return borrowed Py_None on success, or NULL on error. */
+static PyObject *
+textiowrapper_check_write(textio *self)
+{
+    CHECK_ATTACHED(self);
+    CHECK_CLOSED(self);
+
+    if (self->encoder == NULL) {
+        return _unsupported(self->state, "not writable");
+    }
+    return Py_None;
+}
+
 /*[clinic input]
 @critical_section
 _io.TextIOWrapper.write
@@ -1739,11 +1752,8 @@ _io_TextIOWrapper_write_impl(textio *self, PyObject *text)
     int haslf = 0;
     int needflush = 0, text_needflush = 0;
 
-    CHECK_ATTACHED(self);
-    CHECK_CLOSED(self);
-
-    if (self->encoder == NULL) {
-        return _unsupported(self->state, "not writable");
+    if (textiowrapper_check_write(self) == NULL) {
+        return NULL;
     }
 
     Py_INCREF(text);
@@ -1760,6 +1770,17 @@ _io_TextIOWrapper_write_impl(textio *self, PyObject *text)
         Py_DECREF(text);
         if (newtext == NULL)
             return NULL;
+        if (!PyUnicode_Check(newtext)) {
+            PyErr_Format(PyExc_TypeError,
+                         "replace() should return a str, not %T", newtext);
+            Py_DECREF(newtext);
+            return NULL;
+        }
+        /* replace() may have closed, detached or reinitialized the stream. */
+        if (textiowrapper_check_write(self) == NULL) {
+            Py_DECREF(newtext);
+            return NULL;
+        }
         text = newtext;
     }