https://github.com/python/cpython/commit/6a48876b486ce68683d1a61e93685a585974d12b
commit: 6a48876b486ce68683d1a61e93685a585974d12b
branch: 3.14
author: Jelle Zijlstra <[email protected]>
committer: JelleZijlstra <[email protected]>
date: 2026-10-24T13:43:55-07:00
summary:
[3.14] gh-159739: Validate newline replacement in TextIOWrapper.write
(GH-160191) (#160487)
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 c6796a1268292bca0711b05557130ba3406f9850)
files:
A Misc/NEWS.d/next/Library/2026-10-21-21-26-10.gh-issue-159739.k8uuGe.rst
M Lib/test/test_io.py
M Modules/_io/textio.c
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index cbec18fb403632..caf2ae460cf41d 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -4115,6 +4115,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
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 0125e5e5a2d2ba..6dd51cc3fbc521 100644
--- a/Modules/_io/textio.c
+++ b/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;
}
_______________________________________________
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]
