fix: Prevent undefined behaviour from malicious AsyncRead impl (#2030)

`AsyncRead` is safe to implement but can be implemented so that it
reports that it read more bytes than it actually did. `poll_read_buf` on
the other head implicitly trusts that the returned length is actually
correct which makes it possible to advance the buffer past what has
actually been initialized.

An alternative fix could be to avoid the panic and instead advance by
`n.min(b.len())`

tokio/src/io/async_read.rs

@@ -123,7 +123,9 @@ pub trait AsyncRead {
                 // Convert to `&mut [u8]`
                 let b = &mut *(b as *mut [MaybeUninit<u8>] as *mut [u8]);
 
-                ready!(self.poll_read(cx, b))?
+                let n = ready!(self.poll_read(cx, b))?;
+                assert!(n <= b.len(), "Bad AsyncRead implementation, more bytes were reported as read than the buffer can hold");
+                n
             };
 
             buf.advance_mut(n);

tokio/tests/io_read.rs

@@ -36,3 +36,25 @@ async fn read() {
     assert_eq!(n, 11);
     assert_eq!(buf[..], b"hello world"[..]);
 }
+
+struct BadAsyncRead;
+
+impl AsyncRead for BadAsyncRead {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        for b in &mut *buf {
+            *b = b'a';
+        }
+        Poll::Ready(Ok(buf.len() * 2))
+    }
+}
+
+#[tokio::test]
+#[should_panic]
+async fn read_buf_bad_async_read() {
+    let mut buf = Vec::with_capacity(10);
+    BadAsyncRead.read_buf(&mut buf).await.unwrap();
+}