Use Pin<Box<S>> in BodyStream and SizedStream (#1328)

Fixes #1321

A better fix would be to change `MessageBody` to take a `Pin<&mut
Self>`, rather than a `Pin<&mut Self>`. This will avoid requiring the
use of `Box` for all consumers by allowing the caller to determine how
to pin the `MessageBody` implementation (e.g. via stack pinning).

However, doing so is a breaking change that will affect every user of
`MessageBody`. By pinning the inner stream ourselves, we can fix the
undefined behavior without breaking the API.

I've included @sebzim4500's reproduction case as a new test case.
However, due to the nature of undefined behavior, this could pass (and
not segfault) even if underlying issue were to regress.

Unfortunately, until rust-lang/unsafe-code-guidelines#148 is resolved,
it's not even possible to write a Miri test that will pass when the bug
is fixed.

Co-authored-by: Yuki Okushi <huyuumi.dev@gmail.com>

actix-http/src/body.rs

@@ -361,10 +361,8 @@ impl MessageBody for String {
 
 /// Type represent streaming body.
 /// Response does not contain `content-length` header and appropriate transfer encoding is used.
-#[pin_project]
 pub struct BodyStream<S, E> {
-    #[pin]
-    stream: S,
+    stream: Pin<Box<S>>,
     _t: PhantomData<E>,
 }
 
@@ -375,7 +373,7 @@ where
 {
     pub fn new(stream: S) -> Self {
         BodyStream {
-            stream,
+            stream: Box::pin(stream),
             _t: PhantomData,
         }
     }
@@ -396,7 +394,7 @@ where
     /// ended on a zero-length chunk, but rather proceed until the underlying
     /// [`Stream`] ends.
     fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, Error>>> {
-        let mut stream = unsafe { Pin::new_unchecked(self) }.project().stream;
+        let mut stream = self.stream.as_mut();
         loop {
             return Poll::Ready(match ready!(stream.as_mut().poll_next(cx)) {
                 Some(Ok(ref bytes)) if bytes.is_empty() => continue,
@@ -408,11 +406,9 @@ where
 
 /// Type represent streaming body. This body implementation should be used
 /// if total size of stream is known. Data get sent as is without using transfer encoding.
-#[pin_project]
 pub struct SizedStream<S> {
     size: u64,
-    #[pin]
-    stream: S,
+    stream: Pin<Box<S>>,
 }
 
 impl<S> SizedStream<S>
@@ -420,7 +416,7 @@ where
     S: Stream<Item = Result<Bytes, Error>>,
 {
     pub fn new(size: u64, stream: S) -> Self {
-        SizedStream { size, stream }
+        SizedStream { size, stream: Box::pin(stream) }
     }
 }
 
@@ -438,7 +434,7 @@ where
     /// ended on a zero-length chunk, but rather proceed until the underlying
     /// [`Stream`] ends.
     fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, Error>>> {
-        let mut stream = unsafe { Pin::new_unchecked(self) }.project().stream;
+        let mut stream = self.stream.as_mut();
         loop {
             return Poll::Ready(match ready!(stream.as_mut().poll_next(cx)) {
                 Some(Ok(ref bytes)) if bytes.is_empty() => continue,

tests/test_weird_poll.rs

@@ -0,0 +1,26 @@
+// Regression test for #/1321
+
+use futures::task::{noop_waker, Context};
+use futures::stream::once;
+use actix_http::body::{MessageBody, BodyStream};
+use bytes::Bytes;
+
+#[test]
+fn weird_poll() {
+    let (sender, receiver) = futures::channel::oneshot::channel();
+    let mut body_stream = Ok(BodyStream::new(once(async {
+        let x = Box::new(0);
+        let y = &x;
+        receiver.await.unwrap();
+        let _z = **y;
+        Ok::<_, ()>(Bytes::new())
+    })));
+
+    let waker = noop_waker();
+    let mut context = Context::from_waker(&waker);
+
+    let _ = body_stream.as_mut().unwrap().poll_next(&mut context);
+    sender.send(()).unwrap();
+    let _ = std::mem::replace(&mut body_stream, Err([0; 32])).unwrap().poll_next(&mut context);
+}
+