Fix use-after-free in async write causing corruption with large payloads (#392)

In _PyAsyncIoStreamProtocol.write_loop(), memoryview objects pointing to
C++ message memory were passed directly to transport.write(). Since
transport.write() is non-blocking and only queues data for later
transmission, the memoryview could reference freed memory after
fulfill() was called.

This caused message corruption when pipelining RPC calls with payloads
larger than ~4000 bytes, as the C++ message memory would be freed before
asyncio had a chance to transmit the data.

The fix copies the data to Python bytes objects before passing to
transport.write(), ensuring the data remains valid until asyncio
transmits it.

Includes regression test that verifies large payload integrity with both
sequential and pipelined RPC calls.

capnp/lib/capnp.pyx

@@ -17,6 +17,7 @@ from builtins import memoryview as BuiltinsMemoryview
 from cpython cimport array, Py_buffer, PyObject_CheckBuffer
 from cpython.buffer cimport PyBUF_SIMPLE, PyBUF_WRITABLE, PyBUF_WRITE, PyBUF_READ, PyBUF_CONTIG_RO
 from cpython.memoryview cimport PyMemoryView_FromMemory
+from cpython.bytes cimport PyBytes_FromStringAndSize
 from cpython.exc cimport PyErr_Clear
 from cython.operator cimport dereference as deref
 from libc.stdlib cimport malloc, free
@@ -2699,8 +2700,13 @@ cdef class _PyAsyncIoStreamProtocol(DummyBaseClass, asyncio.BufferedProtocol):
         cdef const ArrayPtr[const uint8_t]* piece
         for i in range(self.write_index, self.write_pieces.size()):
             piece = &self.write_pieces[i]
-            view = PyMemoryView_FromMemory(<char*>piece.begin(), piece.size(), PyBUF_READ)
-            self.transport.write(view)
+            # Copy data to Python bytes to avoid use-after-free.
+            # transport.write() is non-blocking and buffers data asynchronously.
+            # The memoryview would point to C++ memory that gets freed when
+            # fulfill() is called below, but asyncio may not have sent the data
+            # yet, causing memory corruption with large payloads.
+            data = PyBytes_FromStringAndSize(<char*>piece.begin(), piece.size())
+            self.transport.write(data)
             if self.write_paused:
                 self.write_index = i+1
                 break