Fix unsoundness issues #106 and #107

async-stream-impl/src/lib.rs

@@ -120,10 +120,19 @@ impl VisitMut for Scrub<'_> {
 
                 // let ident = &self.yielder;
 
-                *i = if self.is_try {
-                    syn::parse_quote! { __yield_tx.send(::core::result::Result::Ok(#value_expr)).await }
+                let yield_expr = if self.is_try {
+                    quote! { __yield_tx.send(::core::result::Result::Ok(#value_expr)).await }
                 } else {
-                    syn::parse_quote! { __yield_tx.send(#value_expr).await }
+                    quote! { __yield_tx.send(#value_expr).await }
+                };
+                *i = syn::parse_quote! {
+                    {
+                        #[allow(unreachable_code)]
+                        if false {
+                            break 'check_stream_scope (loop {});
+                        }
+                        #yield_expr
+                    }
                 };
             }
             syn::Expr::Try(try_expr) => {
@@ -225,8 +234,10 @@ pub fn stream_inner(input: TokenStream) -> TokenStream {
     quote!({
         let (mut __yield_tx, __yield_rx) = unsafe { #crate_path::__private::yielder::pair() };
         #crate_path::__private::AsyncStream::new(__yield_rx, async move {
-            #dummy_yield
-            #(#stmts)*
+            'check_stream_scope: {
+                #dummy_yield
+                #(#stmts)*
+            }
         })
     })
     .into()
@@ -259,8 +270,10 @@ pub fn try_stream_inner(input: TokenStream) -> TokenStream {
     quote!({
         let (mut __yield_tx, __yield_rx) = unsafe { #crate_path::__private::yielder::pair() };
         #crate_path::__private::AsyncStream::new(__yield_rx, async move {
-            #dummy_yield
-            #(#stmts)*
+            'check_stream_scope: {
+                #dummy_yield
+                #(#stmts)*
+            }
         })
     })
     .into()

async-stream/src/yielder.rs

@@ -7,7 +7,7 @@ use std::task::{Context, Poll};
 
 #[derive(Debug)]
 pub struct Sender<T> {
-    _p: PhantomData<T>,
+    _p: PhantomData<fn(T) -> T>,
 }
 
 #[derive(Debug)]

async-stream/tests/ui/unsoundness_issue_106.rs

@@ -0,0 +1,31 @@
+use async_stream::stream;
+use futures_util::StreamExt;
+
+use std::pin::pin;
+
+macro_rules! asynk {
+    ($e:expr) => {
+        async { $e }
+    };
+}
+
+#[tokio::main]
+
+async fn main() {
+    pin!(stream! {
+        let yield_42 = asynk!(yield 42_usize);
+        let s = stream! {
+            yield Box::new(12345);
+            yield_42.await; // yield 42 -- wait that's not a Box!?
+        };
+        for await (n, i) in s.enumerate() {
+            println!("Item at index {n}:\n    {i}");
+            // Item at index 0:
+            //     12345
+            // Item at index 1:
+            // Segmentation fault
+        }
+    })
+    .next()
+    .await;
+}

async-stream/tests/ui/unsoundness_issue_106.stderr

@@ -0,0 +1,36 @@
+error[E0767]: use of unreachable label `'check_stream_scope`
+  --> tests/ui/unsoundness_issue_106.rs:15:10
+   |
+15 |       pin!(stream! {
+   |  __________^
+16 | |         let yield_42 = asynk!(yield 42_usize);
+17 | |         let s = stream! {
+18 | |             yield Box::new(12345);
+...  |
+27 | |         }
+28 | |     })
+   | |     ^
+   | |     |
+   | |_____unreachable label `'check_stream_scope`
+   |       unreachable label defined here
+   |
+   = note: labels are unreachable through functions, closures, async blocks and modules
+   = note: this error originates in the macro `$crate::__private::stream_inner` which comes from the expansion of the macro `stream` (in Nightly builds, run with -Z macro-backtrace for more info)
+
+error[E0267]: `break` inside `async` block
+  --> tests/ui/unsoundness_issue_106.rs:15:10
+   |
+8  |           async { $e }
+   |           ----- enclosing `async` block
+...
+15 |       pin!(stream! {
+   |  __________^
+16 | |         let yield_42 = asynk!(yield 42_usize);
+17 | |         let s = stream! {
+18 | |             yield Box::new(12345);
+...  |
+27 | |         }
+28 | |     })
+   | |_____^ cannot `break` inside `async` block
+   |
+   = note: this error originates in the macro `$crate::__private::stream_inner` which comes from the expansion of the macro `stream` (in Nightly builds, run with -Z macro-backtrace for more info)

async-stream/tests/ui/unsoundness_issue_107.rs

@@ -0,0 +1,23 @@
+use async_stream::stream;
+use futures_util::StreamExt;
+
+use std::pin::pin;
+
+#[tokio::main]
+async fn main() {
+    let mut outer = vec![];
+    {
+        let v = vec![0; 10];
+        let v_ref = &v;
+        let mut s = pin!(stream! {
+            for x in v_ref {
+                yield x
+            }
+        });
+        while let Some(x) = s.next().await {
+            outer.push(x);
+        }
+    };
+    // use-after-free
+    println!("{outer:?}"); // […garbage allocator internals…, 0, 0, 0]
+}

async-stream/tests/ui/unsoundness_issue_107.stderr

@@ -0,0 +1,13 @@
+error[E0597]: `v` does not live long enough
+  --> tests/ui/unsoundness_issue_107.rs:11:21
+   |
+10 |         let v = vec![0; 10];
+   |             - binding `v` declared here
+11 |         let v_ref = &v;
+   |                     ^^ borrowed value does not live long enough
+...
+20 |     };
+   |     - `v` dropped here while still borrowed
+21 |     // use-after-free
+22 |     println!("{outer:?}"); // […garbage allocator internals…, 0, 0, 0]
+   |               --------- borrow later used here