1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
//! Mocking helpers for testing with futures::io types.
//!
//! Note that some of this code might be of general use, but for now
//! we're only trying it for testing.

use futures::channel::mpsc;
use futures::io::{AsyncRead, AsyncWrite};
use futures::sink::{Sink, SinkExt};
use futures::stream::Stream;
use std::io::{Error as IoError, ErrorKind, Result as IoResult};
use std::pin::Pin;
use std::task::{Context, Poll};

/// Channel capacity for our internal MPSC channels.
///
/// We keep this intentionally low to make sure that some blocking
/// will occur occur.
const CAPACITY: usize = 4;

/// Maximum size for a queued buffer on a local chunk.
///
/// This size is deliberately weird, to try to find errors.
const CHUNKSZ: usize = 213;

/// Construct a new pair of linked LocalStream objects.
///
/// Any bytes written to one will be readable on the other, and vice
/// versa.  These streams will behave more or less like a socketpair,
/// except without actually going through the operating system.
///
/// Note that this implementation is intended for testing only, and
/// isn't optimized.
pub fn stream_pair() -> (LocalStream, LocalStream) {
    let (w1, r2) = mpsc::channel(CAPACITY);
    let (w2, r1) = mpsc::channel(CAPACITY);
    let s1 = LocalStream {
        w: w1,
        r: r1,
        pending_bytes: Vec::new(),
        tls_cert: None,
    };
    let s2 = LocalStream {
        w: w2,
        r: r2,
        pending_bytes: Vec::new(),
        tls_cert: None,
    };
    (s1, s2)
}

/// One half of a pair of linked streams returned by [`stream_pair`].
//
// Implementation notes: linked streams are made out a pair of mpsc
// channels.  There's one channel for sending bytes in each direction.
// Bytes are sent as IoResult<Vec<u8>>: sending an error causes an error
// to occur on the other side.
pub struct LocalStream {
    /// The writing side of the channel that we use to implement this
    /// stream.
    ///
    /// The reading side is held by the other linked stream.
    w: mpsc::Sender<IoResult<Vec<u8>>>,
    /// The reading side of the channel that we use to implement this
    /// stream.
    ///
    /// The writing side is held by the other linked stream.
    r: mpsc::Receiver<IoResult<Vec<u8>>>,
    /// Bytes that we have read from `r` but not yet delivered.
    pending_bytes: Vec<u8>,
    /// Data about the other side of this stream's fake TLS certificate, if any.
    /// If this is present, I/O operations will fail with an error.
    ///
    /// How this is intended to work: things that return `LocalStream`s that could potentially
    /// be connected to a fake TLS listener should set this field. Then, a fake TLS wrapper
    /// type would clear this field (after checking its contents are as expected).
    ///
    /// FIXME(eta): this is a bit of a layering violation, but it's hard to do otherwise
    pub(crate) tls_cert: Option<Vec<u8>>,
}

/// Helper: pull bytes off the front of `pending_bytes` and put them
/// onto `buf.  Return the number of bytes moved.
fn drain_helper(buf: &mut [u8], pending_bytes: &mut Vec<u8>) -> usize {
    let n_to_drain = std::cmp::min(buf.len(), pending_bytes.len());
    buf[..n_to_drain].copy_from_slice(&pending_bytes[..n_to_drain]);
    pending_bytes.drain(..n_to_drain);
    n_to_drain
}

impl AsyncRead for LocalStream {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<IoResult<usize>> {
        if buf.is_empty() {
            return Poll::Ready(Ok(0));
        }
        if self.tls_cert.is_some() {
            return Poll::Ready(Err(std::io::Error::new(
                std::io::ErrorKind::Other,
                "attempted to treat a TLS stream as non-TLS!",
            )));
        }
        if !self.pending_bytes.is_empty() {
            return Poll::Ready(Ok(drain_helper(buf, &mut self.pending_bytes)));
        }

        match futures::ready!(Pin::new(&mut self.r).poll_next(cx)) {
            Some(Err(e)) => Poll::Ready(Err(e)),
            Some(Ok(bytes)) => {
                self.pending_bytes = bytes;
                let n = drain_helper(buf, &mut self.pending_bytes);
                Poll::Ready(Ok(n))
            }
            None => Poll::Ready(Ok(0)), // This is an EOF
        }
    }
}

impl AsyncWrite for LocalStream {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<IoResult<usize>> {
        if self.tls_cert.is_some() {
            return Poll::Ready(Err(std::io::Error::new(
                std::io::ErrorKind::Other,
                "attempted to treat a TLS stream as non-TLS!",
            )));
        }

        match futures::ready!(Pin::new(&mut self.w).poll_ready(cx)) {
            Ok(()) => (),
            Err(e) => return Poll::Ready(Err(IoError::new(ErrorKind::BrokenPipe, e))),
        }

        let buf = if buf.len() > CHUNKSZ {
            &buf[..CHUNKSZ]
        } else {
            buf
        };
        let len = buf.len();
        match Pin::new(&mut self.w).start_send(Ok(buf.to_vec())) {
            Ok(()) => Poll::Ready(Ok(len)),
            Err(e) => Poll::Ready(Err(IoError::new(ErrorKind::BrokenPipe, e))),
        }
    }
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
        Pin::new(&mut self.w)
            .poll_flush(cx)
            .map_err(|e| IoError::new(ErrorKind::BrokenPipe, e))
    }
    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
        Pin::new(&mut self.w)
            .poll_close(cx)
            .map_err(|e| IoError::new(ErrorKind::Other, e))
    }
}

/// An error generated by [`LocalStream::send_err`].
#[derive(Debug, Clone, Eq, PartialEq)]
#[non_exhaustive]
pub struct SyntheticError;
impl std::error::Error for SyntheticError {}
impl std::fmt::Display for SyntheticError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Synthetic error")
    }
}

impl LocalStream {
    /// Send an error to the other linked local stream.
    ///
    /// When the other stream reads this message, it will generate a
    /// [`std::io::Error`] with the provided `ErrorKind`.
    pub async fn send_err(&mut self, kind: ErrorKind) {
        let _ignore = self.w.send(Err(IoError::new(kind, SyntheticError))).await;
    }
}

#[cfg(test)]
mod test {
    #![allow(clippy::unwrap_used)]
    use super::*;

    use futures::io::{AsyncReadExt, AsyncWriteExt};
    use futures_await_test::async_test;
    use rand::Rng;
    use tor_basic_utils::test_rng::testing_rng;

    #[async_test]
    async fn basic_rw() {
        let (mut s1, mut s2) = stream_pair();
        let mut text1 = vec![0_u8; 9999];
        testing_rng().fill(&mut text1[..]);

        let (v1, v2): (IoResult<()>, IoResult<()>) = futures::join!(
            async {
                for _ in 0_u8..10 {
                    s1.write_all(&text1[..]).await?;
                }
                s1.close().await?;
                Ok(())
            },
            async {
                let mut text2: Vec<u8> = Vec::new();
                let mut buf = [0_u8; 33];
                loop {
                    let n = s2.read(&mut buf[..]).await?;
                    if n == 0 {
                        break;
                    }
                    text2.extend(&buf[..n]);
                }
                for ch in text2[..].chunks(text1.len()) {
                    assert_eq!(ch, &text1[..]);
                }
                Ok(())
            }
        );

        v1.unwrap();
        v2.unwrap();
    }

    #[async_test]
    async fn send_error() {
        let (mut s1, mut s2) = stream_pair();

        let (v1, v2): (IoResult<()>, IoResult<()>) = futures::join!(
            async {
                s1.write_all(b"hello world").await?;
                s1.send_err(ErrorKind::PermissionDenied).await;
                Ok(())
            },
            async {
                let mut buf = [0_u8; 33];
                loop {
                    let n = s2.read(&mut buf[..]).await?;
                    if n == 0 {
                        break;
                    }
                }
                Ok(())
            }
        );

        v1.unwrap();
        let e = v2.err().unwrap();
        assert_eq!(e.kind(), ErrorKind::PermissionDenied);
        let synth = e.into_inner().unwrap();
        assert_eq!(synth.to_string(), "Synthetic error");
    }

    #[async_test]
    async fn drop_reader() {
        let (mut s1, s2) = stream_pair();

        let (v1, v2): (IoResult<()>, IoResult<()>) = futures::join!(
            async {
                for _ in 0_u16..1000 {
                    s1.write_all(&[9_u8; 9999]).await?;
                }
                Ok(())
            },
            async {
                drop(s2);
                Ok(())
            }
        );

        v2.unwrap();
        let e = v1.err().unwrap();
        assert_eq!(e.kind(), ErrorKind::BrokenPipe);
    }
}