From d509daf5e183330f03ca4f4a8212e24bd188a699 Mon Sep 17 00:00:00 2001 From: Seonghun An <53287605+shAn-kor@users.noreply.github.com> Date: Sat, 18 Jul 2026 16:39:23 +0900 Subject: [PATCH] Validate TextIOWrapper chunk size (#8265) * Validate TextIOWrapper chunk size Match CPython for non-positive, oversized, and index-compatible values. Assisted-by: Codex:GPT-5 * Use explicit AssertionError in io snippet Assisted-by: Codex:GPT-5 * Report original type in chunk size overflow Assisted-by: Codex:GPT-5 * Chain io snippet assertion error Assisted-by: Codex:GPT-5 * Match CPython chunk size setter behavior Assisted-by: Codex:GPT-5 * Format io snippet assertion Assisted-by: Codex:GPT-5.6 * Match CPython type name truncation Assisted-by: Codex:GPT-5 * Format non-ASCII chunk size test Assisted-by: Codex:GPT-5 --- crates/vm/src/stdlib/_io.rs | 50 +++++++++---- extra_tests/snippets/stdlib_io.py | 113 ++++++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 12 deletions(-) diff --git a/crates/vm/src/stdlib/_io.rs b/crates/vm/src/stdlib/_io.rs index 6707a254c..c7bbbaf93 100644 --- a/crates/vm/src/stdlib/_io.rs +++ b/crates/vm/src/stdlib/_io.rs @@ -3206,19 +3206,45 @@ mod _io { } #[pygetset(setter, name = "_CHUNK_SIZE")] - fn set_chunksize( - &self, - chunk_size: PySetterValue, - vm: &VirtualMachine, - ) -> PyResult<()> { - let mut textio = self.lock(vm)?; - match chunk_size { - PySetterValue::Assign(chunk_size) => textio.chunk_size = chunk_size, - PySetterValue::Delete => Err(vm.new_attribute_error("cannot delete attribute"))?, + fn set_chunksize(&self, value: PySetterValue, vm: &VirtualMachine) -> PyResult<()> { + { + let textio = self.lock(vm)?; + if vm.is_none(&textio.buffer) { + return Err(vm.new_value_error("underlying buffer has been detached")); + } + } + + let chunk_size: isize = match value { + PySetterValue::Assign(object_value) => { + let integer = object_value.try_index(vm)?; + + integer.try_to_primitive::(vm).map_err(|_| { + let class = object_value.class(); + let type_name = class.name(); + let mut end = type_name.len().min(200); + while !type_name.is_char_boundary(end) { + end -= 1; + } + vm.new_value_error(format!( + "cannot fit '{}' into an index-sized integer", + &type_name[..end] + )) + })? + } + PySetterValue::Delete => { + return Err(vm.new_attribute_error("cannot delete attribute")); + } }; - // TODO: RUSTPYTHON - // Change chunk_size type, validate it manually and throws ValueError if invalid. - // https://github.com/python/cpython/blob/2e9da8e3522764d09f1d6054a2be567e91a30812/Modules/_io/textio.c#L3124-L3143 + + if chunk_size <= 0 { + return Err(vm.new_value_error("a strictly positive integer is required")); + } + + let chunk_size = usize::try_from(chunk_size) + .map_err(|_| vm.new_value_error("a strictly positive integer is required"))?; + + let mut textio = self.lock(vm)?; + textio.chunk_size = chunk_size; Ok(()) } diff --git a/extra_tests/snippets/stdlib_io.py b/extra_tests/snippets/stdlib_io.py index 93c083a90..f17eae5b1 100644 --- a/extra_tests/snippets/stdlib_io.py +++ b/extra_tests/snippets/stdlib_io.py @@ -84,3 +84,116 @@ textio = TextIOWrapper(raw, encoding="utf-8", write_through=True) raw.textio = textio with assert_raises(AttributeError): textio.writelines(["x"]) + +textio = TextIOWrapper(BytesIO()) + +for invalid_chunk_size in (0, -1, 2**100): + try: + textio._CHUNK_SIZE = invalid_chunk_size + except ValueError: + pass + else: + raise AssertionError(f"expected ValueError for {invalid_chunk_size!r}") + +for invalid_chunk_size in (1.5, "4"): + try: + textio._CHUNK_SIZE = invalid_chunk_size + except TypeError: + pass + else: + raise AssertionError(f"expected TypeError for {invalid_chunk_size!r}") + + +class ChunkSize: + def __index__(self): + return 16 + + +textio._CHUNK_SIZE = ChunkSize() +assert textio._CHUNK_SIZE == 16 + + +class OversizedChunkSize: + def __index__(self): + return 2**100 + + +try: + textio._CHUNK_SIZE = OversizedChunkSize() +except ValueError as error: + expected = "cannot fit 'OversizedChunkSize' into an index-sized integer" + if str(error) != expected: + raise AssertionError(f"unexpected error message: {error}") from error +else: + raise AssertionError("expected ValueError for oversized indexable object") + + +def expect_value_error(expected, operation): + try: + operation() + except ValueError as error: + if str(error) != expected: + raise AssertionError(f"unexpected error message: {error}") from error + else: + raise AssertionError(f"expected ValueError: {expected}") + + +class UninitializedChunkSize: + def __init__(self): + self.called = False + + def __index__(self): + self.called = True + return 16 + + +uninitialized_textio = TextIOWrapper.__new__(TextIOWrapper) +uninitialized_chunk_size = UninitializedChunkSize() +expect_value_error( + "I/O operation on uninitialized object", + lambda: setattr(uninitialized_textio, "_CHUNK_SIZE", uninitialized_chunk_size), +) + +if uninitialized_chunk_size.called: + raise AssertionError( + "__index__ should not be called for uninitialized TextIOWrapper" + ) + + +detached_textio = TextIOWrapper(BytesIO()) +detached_textio.detach() +expect_value_error( + "underlying buffer has been detached", + lambda: setattr(detached_textio, "_CHUNK_SIZE", 16), +) +expect_value_error( + "underlying buffer has been detached", + lambda: delattr(detached_textio, "_CHUNK_SIZE"), +) + + +long_type_name = "X" * 250 +LongNamedChunkSize = type( + long_type_name, + (), + {"__index__": lambda self: 2**100}, +) +expect_value_error( + f"cannot fit '{long_type_name[:200]}' into an index-sized integer", + lambda: setattr(textio, "_CHUNK_SIZE", LongNamedChunkSize()), +) + + +non_ascii_type_name = "é" * 250 +NonAsciiNamedChunkSize = type( + non_ascii_type_name, + (), + {"__index__": lambda self: 2**100}, +) +truncated_non_ascii_type_name = non_ascii_type_name.encode("utf-8")[:200].decode( + "utf-8" +) +expect_value_error( + f"cannot fit '{truncated_non_ascii_type_name}' into an index-sized integer", + lambda: setattr(textio, "_CHUNK_SIZE", NonAsciiNamedChunkSize()), +)