From 391c7d3ecffb94cc1618b806a2f95ab0b4a5b157 Mon Sep 17 00:00:00 2001 From: Nacl-HCl <308349766+Nacl-HCl@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:31:32 +0000 Subject: [PATCH] Fix send_all() silently truncating typed memoryviews after partial writes SocketStream.send_all() and POSIX FdStream.send_all() used the byte counts returned by send()/os.write() as memoryview element offsets. For a contiguous memoryview with itemsize > 1, a partial write could skip data and return successfully without sending the full buffer. Cast non-empty views to a one-dimensional byte view before slicing, so offsets and write return values are both in bytes. Empty views keep the previous early-return path because cast() rejects strided views, even empty ones. Fixes #3543 --- newsfragments/3543.bugfix.rst | 2 ++ src/trio/_highlevel_socket.py | 15 ++++++--- src/trio/_tests/test_highlevel_socket.py | 41 ++++++++++++++++++++++++ src/trio/_tests/test_unix_pipes.py | 33 +++++++++++++++++++ src/trio/_unix_pipes.py | 37 ++++++++++++--------- 5 files changed, 108 insertions(+), 20 deletions(-) create mode 100644 newsfragments/3543.bugfix.rst diff --git a/newsfragments/3543.bugfix.rst b/newsfragments/3543.bugfix.rst new file mode 100644 index 000000000..10b10209c --- /dev/null +++ b/newsfragments/3543.bugfix.rst @@ -0,0 +1,2 @@ +Fix ``SocketStream.send_all()`` and POSIX ``FdStream.send_all()`` silently truncating typed +memoryviews (``itemsize > 1``) when ``send()``/``os.write()`` performs a partial write. diff --git a/src/trio/_highlevel_socket.py b/src/trio/_highlevel_socket.py index 142ab11e0..db6d9cf78 100644 --- a/src/trio/_highlevel_socket.py +++ b/src/trio/_highlevel_socket.py @@ -117,11 +117,16 @@ async def send_all(self, data: bytes | bytearray | memoryview) -> None: raise trio.ClosedResourceError("socket was already closed") await trio.lowlevel.checkpoint() return - total_sent = 0 - while total_sent < len(data): - with data[total_sent:] as remaining: - sent = await self.socket.send(remaining) - total_sent += sent + # send() returns a byte count, so the buffer must be + # sliced in bytes: cast typed views (itemsize > 1) to a + # one-dimensional byte view first, otherwise a partial + # write silently skips data (gh-3543). + with data.cast("B") as data: + total_sent = 0 + while total_sent < len(data): + with data[total_sent:] as remaining: + sent = await self.socket.send(remaining) + total_sent += sent async def wait_send_all_might_not_block(self) -> None: with self._send_conflict_detector: diff --git a/src/trio/_tests/test_highlevel_socket.py b/src/trio/_tests/test_highlevel_socket.py index ff19e3367..8bb7dcee8 100644 --- a/src/trio/_tests/test_highlevel_socket.py +++ b/src/trio/_tests/test_highlevel_socket.py @@ -119,6 +119,47 @@ async def receiver() -> None: assert await b.receive_some(10) == b"" +async def test_SocketStream_send_all_typed_memoryview( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Regression test for gh-3543: send_all() must not silently truncate a + # typed memoryview (itemsize > 1) when send() performs partial writes. + a_sock, b_sock = tsocket.socketpair() + with a_sock, b_sock: + a = SocketStream(a_sock) + b = SocketStream(b_sock) + + # Force partial writes: at most 3 bytes per send() call. + real_send = a_sock.send + + def limited_send(buf: bytes | bytearray | memoryview, *args: object) -> int: + return real_send(memoryview(buf)[:3]) # type: ignore[return-value] + + monkeypatch.setattr(a_sock, "send", limited_send) + + payload = bytes(range(256)) * 64 + received = bytearray() + + async def sender() -> None: + with memoryview(payload).cast("H") as typed: + assert typed.itemsize == 2 + await a.send_all(typed) + await a.send_eof() + + async def receiver() -> None: + while True: + chunk = await b.receive_some(65536) + if not chunk: + break + received.extend(chunk) + + async with _core.open_nursery() as nursery: + nursery.start_soon(sender) + nursery.start_soon(receiver) + + assert bytes(received) == payload + + async def fill_stream(s: SocketStream) -> None: async def sender() -> None: while True: diff --git a/src/trio/_tests/test_unix_pipes.py b/src/trio/_tests/test_unix_pipes.py index f81004049..a5d227da6 100644 --- a/src/trio/_tests/test_unix_pipes.py +++ b/src/trio/_tests/test_unix_pipes.py @@ -66,6 +66,39 @@ async def test_send_pipe() -> None: os.close(r) +async def test_send_all_typed_memoryview(monkeypatch: pytest.MonkeyPatch) -> None: + # Regression test for gh-3543: FdStream.send_all() must not silently + # truncate a typed memoryview (itemsize > 1) on partial writes. + write, read = await make_pipe() + + # Force partial writes: at most 3 bytes per os.write() call. + real_write = os.write + + def limited_write(fd: int, buf: bytes | bytearray | memoryview) -> int: + return real_write(fd, memoryview(buf)[:3]) + + monkeypatch.setattr(os, "write", limited_write) + + payload = bytes(range(256)) * 64 + with memoryview(payload).cast("H") as typed: + assert typed.itemsize == 2 + await write.send_all(typed) + # Empty views (even strided ones, which cast("B") would reject) are + # still accepted. + await write.send_all(b"") + await write.send_all(memoryview(b"")[::2]) + await write.aclose() + + received = bytearray() + while True: + chunk = await read.receive_some(65536) + if not chunk: + break + received.extend(chunk) + await read.aclose() + assert bytes(received) == payload + + async def test_receive_pipe() -> None: r, w = os.pipe() async with FdStream(r) as recv: diff --git a/src/trio/_unix_pipes.py b/src/trio/_unix_pipes.py index 598feb61e..67c4cbcb7 100644 --- a/src/trio/_unix_pipes.py +++ b/src/trio/_unix_pipes.py @@ -134,23 +134,30 @@ async def send_all(self, data: bytes | bytearray | memoryview) -> None: if self._fd_holder.closed: raise trio.ClosedResourceError("file was already closed") await trio.lowlevel.checkpoint() - length = len(data) # adapted from the SocketStream code with memoryview(data) as view: - sent = 0 - while sent < length: - with view[sent:] as remaining: - try: - sent += os.write(self._fd_holder.fd, remaining) - except BlockingIOError: - await trio.lowlevel.wait_writable(self._fd_holder.fd) - except OSError as e: - if e.errno == errno.EBADF: - raise trio.ClosedResourceError( - "file was already closed", - ) from None - else: - raise trio.BrokenResourceError from e + if not view: + return + # os.write() returns a byte count, so the buffer must be + # sliced in bytes: cast typed views (itemsize > 1) to a + # one-dimensional byte view first, otherwise a partial write + # silently skips data (gh-3543). The empty case is handled + # above because cast() rejects strided views, even empty ones. + with view.cast("B") as view: + sent = 0 + while sent < len(view): + with view[sent:] as remaining: + try: + sent += os.write(self._fd_holder.fd, remaining) + except BlockingIOError: + await trio.lowlevel.wait_writable(self._fd_holder.fd) + except OSError as e: + if e.errno == errno.EBADF: + raise trio.ClosedResourceError( + "file was already closed", + ) from None + else: + raise trio.BrokenResourceError from e async def wait_send_all_might_not_block(self) -> None: with self._send_conflict_detector: