gh-159739: Validate newline replacement in TextIOWrapper.write

Lib/test/test_io/test_textio.py

@@ -1448,6 +1448,41 @@ def _to_memoryview(buf):
 class CTextIOWrapperTest(TextIOWrapperTest, CTestCase):
     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_chunk_size(self):
         t = self.TextIOWrapper(self.BytesIO(), encoding="utf-8")
         self.assertGreater(t._CHUNK_SIZE, 0)

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

@@ -1730,6 +1730,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
@@ -1747,11 +1760,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);
@@ -1768,6 +1778,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;
     }