https://github.com/python/cpython/commit/c6796a1268292bca0711b05557130ba3406f9850
commit: c6796a1268292bca0711b05557130ba3406f9850
branch: main
author: Jelle Zijlstra <[email protected]>
committer: JelleZijlstra <[email protected]>
date: 2026-10-22T14:17:22-07:00
summary:
gh-159739: Validate newline replacement in TextIOWrapper.write (#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.
files:
A Misc/NEWS.d/next/Library/2026-10-21-21-26-10.gh-issue-159739.k8uuGe.rst
M Lib/test/test_io/test_textio.py
M Modules/_io/textio.c
diff --git a/Lib/test/test_io/test_textio.py b/Lib/test/test_io/test_textio.py
index 65b0812f071e21..04867f113b05a9 100644
--- a/Lib/test/test_io/test_textio.py
+++ b/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)
diff --git
a/Misc/NEWS.d/next/Library/2026-10-21-21-26-10.gh-issue-159739.k8uuGe.rst
b/Misc/NEWS.d/next/Library/2026-10-21-21-26-10.gh-issue-159739.k8uuGe.rst
new file mode 100644
index 00000000000000..24dd0b7c7776e3
--- /dev/null
+++ b/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.
diff --git a/Modules/_io/textio.c b/Modules/_io/textio.c
index f33ab46b10fd2e..7523b516f49eb5 100644
--- a/Modules/_io/textio.c
+++ b/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;
}
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
