1
//! Wrap [tor_cell::chancell::codec::ChannelCodec] for use with the asynchronous_codec
2
//! crate.
3

            
4
use digest::Digest;
5
use tor_bytes::Reader;
6
use tor_cell::chancell::{
7
    AnyChanCell, ChanCell, ChanCmd, ChanMsg, codec,
8
    msg::{self, AnyChanMsg},
9
};
10
use tor_error::internal;
11
use tor_llcrypto as ll;
12

            
13
use bytes::BytesMut;
14

            
15
use crate::{channel::msg::LinkVersion, util::err::Error as ChanError};
16

            
17
use super::{ChannelType, msg::MessageFilter};
18

            
19
/// An authentication rolling digest value.
20
pub(crate) type AuthLogDigest = [u8; 32];
21
/// The rolling digest for the bytes sent on a channel. (Received by the responder).
22
#[derive(Debug, PartialEq)]
23
pub(crate) struct ClogDigest(AuthLogDigest);
24
/// The rolling digest for the bytes received on a channel. (Sent by the responder).
25
#[derive(Debug, PartialEq)]
26
pub(crate) struct SlogDigest(AuthLogDigest);
27

            
28
impl ClogDigest {
29
    /// Constructor from a `AuthLogDigest`
30
    pub(crate) fn new(digest: AuthLogDigest) -> Self {
31
        Self(digest)
32
    }
33
}
34

            
35
impl SlogDigest {
36
    /// Constructor from a `AuthLogDigest`
37
4
    pub(crate) fn new(digest: AuthLogDigest) -> Self {
38
4
        Self(digest)
39
4
    }
40
}
41

            
42
impl AsRef<[u8]> for ClogDigest {
43
    fn as_ref(&self) -> &[u8] {
44
        &self.0
45
    }
46
}
47
impl AsRef<[u8]> for SlogDigest {
48
    fn as_ref(&self) -> &[u8] {
49
        &self.0
50
    }
51
}
52

            
53
/// Channel cell handler which is always in three state.
54
///
55
/// This ALWAYS starts the handler at New. This can only be constructed from a [ChannelType] which
56
/// forces it to start at New.
57
///
58
/// From the New state, it will automatically transition to the right state as information is
59
/// attached to it (ex: link protocol version).
60
pub(crate) enum ChannelCellHandler {
61
    /// When a network connection opens to another endpoint, the channel is considered "New" and
62
    /// so we use this handler to start the handshake.
63
    New(NewChannelHandler),
64
    /// We opened and negotiated a VERSIONS cell. If successful, we transition to this cell handler
65
    /// with sole purpose to handle the handshake phase.
66
    Handshake(HandshakeChannelHandler),
67
    /// Once the handshake is successful, the channel is Open and we use this handler.
68
    Open(OpenChannelHandler),
69
}
70

            
71
/// This is the only way to construct a ChannelCellHandler, from the channel type which will always
72
/// start the handler at the New state.
73
impl From<super::ChannelType> for ChannelCellHandler {
74
97
    fn from(ty: ChannelType) -> Self {
75
97
        Self::New(ty.into())
76
97
    }
77
}
78

            
79
impl ChannelCellHandler {
80
    /// Return the [`ChannelType`] of the inner handler.
81
51
    pub(crate) fn channel_type(&self) -> ChannelType {
82
51
        match self {
83
            Self::New(h) => h.channel_type,
84
            Self::Handshake(h) => h.channel_type(),
85
51
            Self::Open(h) => h.channel_type(),
86
        }
87
51
    }
88

            
89
    /// Set link protocol for this channel cell handler. This transition the handler into the
90
    /// handshake handler state.
91
    ///
92
    /// An error is returned if the current handler is NOT the New one or if the link version is
93
    /// unknown.
94
93
    pub(crate) fn set_link_version(&mut self, link_version: u16) -> Result<(), ChanError> {
95
93
        let Self::New(new_handler) = self else {
96
            return Err(ChanError::Bug(internal!(
97
                "Setting link protocol without a new handler",
98
            )));
99
        };
100
93
        *self = Self::Handshake(new_handler.next_handler(link_version.try_into()?));
101
93
        Ok(())
102
93
    }
103

            
104
    /// This transition into the open handler state.
105
    ///
106
    /// An error is returned if the current handler is NOT the Handshake one.
107
63
    pub(crate) fn set_open(&mut self) -> Result<(), ChanError> {
108
63
        let Self::Handshake(handler) = self else {
109
            return Err(ChanError::Bug(internal!(
110
                "Setting open without a handshake handler"
111
            )));
112
        };
113
63
        *self = Self::Open(handler.next_handler());
114
63
        Ok(())
115
63
    }
116

            
117
    /// Mark this handler as authenticated.
118
    ///
119
    /// This can only happen during the Handshake process as a New handler can't be authenticated
120
    /// from the start and an Open handler can only be opened after authentication.
121
    pub(crate) fn set_authenticated(&mut self) -> Result<(), ChanError> {
122
        let Self::Handshake(handler) = self else {
123
            return Err(ChanError::Bug(internal!(
124
                "Setting authenticated without a handshake handler"
125
            )));
126
        };
127
        handler.set_authenticated();
128
        Ok(())
129
    }
130

            
131
    /// The digest of bytes sent on this channel.
132
    ///
133
    /// This should only ever be called once as it consumes the send log.
134
    ///
135
    /// This will return an error if one of:
136
    /// - The channel is not recording the send log.
137
    /// - The send log digest has already been taken.
138
    /// - This cell handler is not using a handshake handler.
139
6
    pub(crate) fn take_send_log_digest(&mut self) -> Result<AuthLogDigest, ChanError> {
140
6
        if let Self::Handshake(handler) = self {
141
6
            handler
142
6
                .take_send_log_digest()
143
6
                .ok_or(ChanError::Bug(internal!(
144
6
                    "No send log digest on channel, or already taken"
145
6
                )))
146
        } else {
147
            Err(ChanError::Bug(internal!(
148
                "Getting send log digest without a handshake handler"
149
            )))
150
        }
151
6
    }
152

            
153
    /// The digest of bytes received on this channel.
154
    ///
155
    /// This should only ever be called once as it consumes the receive log.
156
    ///
157
    /// This will return `None` if one of:
158
    /// - The channel is not recording the receive log.
159
    /// - The receive log digest has already been taken.
160
    /// - This cell handler is not using a handshake handler.
161
10
    pub(crate) fn take_recv_log_digest(&mut self) -> Result<AuthLogDigest, ChanError> {
162
10
        if let Self::Handshake(handler) = self {
163
10
            handler
164
10
                .take_recv_log_digest()
165
10
                .ok_or(ChanError::Bug(internal!(
166
10
                    "No recv log digest on channel, or already taken"
167
10
                )))
168
        } else {
169
            Err(ChanError::Bug(internal!(
170
                "Getting recv log digest without a handshake handler"
171
            )))
172
        }
173
10
    }
174
}
175

            
176
// Security Consideration.
177
//
178
// Here is an explanation on why AnyChanCell is used as Item in the Handshake and Open handler and
179
// thus the higher level ChannelCellHandler.
180
//
181
// Technically, we could use a restricted message set and so the decoding and encoding wouldn't do
182
// anything if the cell/data was not part of that set.
183
//
184
// However, with relay and client, we have multiple channel types which means we have now a lot
185
// more sets of restricted message (see msg.rs) and each of them are per link protocol version, per
186
// stage of the channel opening process and per direction (inbound or outbound).
187
//
188
// To go around this, we use [MessageFilter] in order to decode on the specific restricted message
189
// set but still return a [AnyChanCell].
190
//
191
// If someone wants to contribute a more elegant solution that wouldn't require us to duplicate
192
// code for each restricted message set, by all means, go for it :).
193

            
194
impl asynchronous_codec::Decoder for ChannelCellHandler {
195
    type Item = AnyChanCell;
196
    type Error = ChanError;
197

            
198
1078
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
199
1078
        match self {
200
206
            Self::New(c) => c
201
206
                .decode(src)
202
206
                .map(|opt| opt.map(|msg| ChanCell::new(None, msg.into()))),
203
860
            Self::Handshake(c) => c.decode(src),
204
12
            Self::Open(c) => c.decode(src),
205
        }
206
1078
    }
207
}
208

            
209
impl asynchronous_codec::Encoder for ChannelCellHandler {
210
    type Item<'a> = AnyChanCell;
211
    type Error = ChanError;
212

            
213
140
    fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
214
140
        match self {
215
78
            Self::New(c) => {
216
                // The new handler pins the only possible message to be a Versions. That is why we
217
                // extract it here and validate before else we can't pass Item to encode().
218
78
                let AnyChanMsg::Versions(versions) = item.into_circid_and_msg().1 else {
219
                    return Err(Self::Error::HandshakeProto(
220
                        "Non VERSIONS cell for new handler".into(),
221
                    ));
222
                };
223
78
                c.encode(versions, dst)
224
            }
225
56
            Self::Handshake(c) => c.encode(item, dst),
226
6
            Self::Open(c) => c.encode(item, dst),
227
        }
228
140
    }
229
}
230

            
231
/// A new channel handler used when a channel is created but before the handshake meaning there is no
232
/// link protocol version yet associated with it.
233
///
234
/// This handler only handles the VERSIONS cell.
235
pub(crate) struct NewChannelHandler {
236
    /// The channel type for this handler.
237
    channel_type: ChannelType,
238
    /// The digest of bytes sent on this channel.
239
    ///
240
    /// Will be used for the SLOG or CLOG of the AUTHENTICATE cell.
241
    send_log: Option<ll::d::Sha256>,
242
    /// The digest of bytes received on this channel.
243
    ///
244
    /// Will be used for the SLOG or CLOG of the AUTHENTICATE cell.
245
    recv_log: Option<ll::d::Sha256>,
246
}
247

            
248
impl NewChannelHandler {
249
    /// Return a handshake handler ready for the given link protocol.
250
93
    fn next_handler(&mut self, link_version: LinkVersion) -> HandshakeChannelHandler {
251
93
        HandshakeChannelHandler::new(self, link_version)
252
93
    }
253
}
254

            
255
impl From<ChannelType> for NewChannelHandler {
256
97
    fn from(channel_type: ChannelType) -> Self {
257
97
        match channel_type {
258
74
            ChannelType::ClientInitiator => Self {
259
74
                channel_type,
260
74
                send_log: None,
261
74
                recv_log: None,
262
74
            },
263
            // Relay responder might not need clog/slog but that is fine. We don't know until the
264
            // end of the handshake.
265
23
            ChannelType::RelayInitiator | ChannelType::RelayResponder { .. } => Self {
266
23
                channel_type,
267
23
                send_log: Some(ll::d::Sha256::new()),
268
23
                recv_log: Some(ll::d::Sha256::new()),
269
23
            },
270
        }
271
97
    }
272
}
273

            
274
impl asynchronous_codec::Decoder for NewChannelHandler {
275
    type Item = msg::Versions;
276
    type Error = ChanError;
277

            
278
206
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
279
        // NOTE: Until the body can be extracted from src buffer, it MUST NOT be modified as in
280
        // advanced with the Buf trait or modified in any ways. Reason is that we can realize we
281
        // don't have enough bytes in the src buffer for the expected body length from the header
282
        // so we have to leave the src buffer untouched and wait for more bytes.
283

            
284
        // See tor-spec, starting a handshake, all cells are variable length so the first 5 bytes
285
        // are: CircId as u16, Command as u8, Length as u16 totalling 5 bytes.
286
        const HEADER_SIZE: usize = 5;
287

            
288
        // Below this amount, this is not a valid cell we can decode. This is important because we
289
        // can get an empty buffer in normal circumstances (see how Framed work) and so we have to
290
        // return that we weren't able to decode and thus no Item.
291
206
        if src.len() < HEADER_SIZE {
292
128
            return Ok(None);
293
78
        }
294

            
295
        // Get the CircID and Command from the header. This is safe due to the header size check
296
        // above.
297
78
        let circ_id = u16::from_be_bytes([src[0], src[1]]);
298
78
        if circ_id != 0 {
299
2
            return Err(Self::Error::HandshakeProto(
300
2
                "Invalid CircID in variable cell".into(),
301
2
            ));
302
76
        }
303

            
304
        // We are only expecting these specific commands. We have to do this by hand here as after
305
        // that we can use a proper codec.
306
76
        let cmd = ChanCmd::from(src[2]);
307
76
        if cmd != ChanCmd::VERSIONS {
308
            return Err(Self::Error::HandshakeProto(format!(
309
                "Invalid command {cmd} variable cell, expected a VERSIONS."
310
            )));
311
76
        }
312

            
313
        // Get the body length now from the next two bytes. This is still safe due to the first
314
        // header size check at the start.
315
76
        let body_len = u16::from_be_bytes([src[3], src[4]]) as usize;
316

            
317
        // See https://gitlab.torproject.org/tpo/core/tor/-/issues/10365. The gist is that because
318
        // version numbers are u16, an odd payload would mean we have a trailing byte that is
319
        // unused which shouldn't be and because we don't expect not controlled that byte, as maxi
320
        // precaution, we don't allow.
321
76
        if body_len % 2 == 1 {
322
            return Err(Self::Error::HandshakeProto(
323
                "VERSIONS cell body length is odd. Rejecting.".into(),
324
            ));
325
76
        }
326

            
327
        // Make sure we have enough bytes in our payload.
328
76
        let wanted_bytes = HEADER_SIZE + body_len;
329
76
        if src.len() < wanted_bytes {
330
            // We don't haven't received enough data to decode the expected length from the header
331
            // so return no Item.
332
            //
333
            // IMPORTANT: The src buffer here can't be advance before reaching this check.
334
            return Ok(None);
335
76
        }
336
        // Extract the exact data we will be looking at.
337
76
        let mut data = src.split_to(wanted_bytes);
338

            
339
        // Update the receive log digest with the entire cell up to the end of the payload hence the
340
        // data we are looking at (and not the whole source). Even on error, this doesn't matter
341
        // because if decoding fails, the channel is closed.
342
76
        if let Some(recv_log) = self.recv_log.as_mut() {
343
16
            recv_log.update(&data);
344
66
        }
345

            
346
        // Get the actual body from the data.
347
76
        let body = data.split_off(HEADER_SIZE).freeze();
348
76
        let mut reader = Reader::from_bytes(&body);
349

            
350
        // Decode the VERSIONS.
351
76
        let cell = msg::Versions::decode_from_reader(cmd, &mut reader)
352
76
            .map_err(|e| Self::Error::from_bytes_err(e, "new cell handler"))?;
353
76
        Ok(Some(cell))
354
206
    }
355
}
356

            
357
impl asynchronous_codec::Encoder for NewChannelHandler {
358
    type Item<'a> = msg::Versions;
359
    type Error = ChanError;
360

            
361
78
    fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
362
78
        let encoded_bytes = item
363
78
            .encode_for_handshake()
364
78
            .map_err(|e| Self::Error::from_bytes_enc(e, "new cell handler"))?;
365
        // Update the send log digest.
366
78
        if let Some(send_log) = self.send_log.as_mut() {
367
17
            send_log.update(&encoded_bytes);
368
67
        }
369
        // Special encoding for the VERSIONS cell.
370
78
        dst.extend_from_slice(&encoded_bytes);
371
78
        Ok(())
372
78
    }
373
}
374

            
375
/// The handshake channel handler which is used to decode and encode cells onto a channel that is
376
/// handshaking with an endpoint.
377
pub(crate) struct HandshakeChannelHandler {
378
    /// Message filter used to allow or not a certain message.
379
    filter: MessageFilter,
380
    /// The cell codec that we'll use to encode and decode our cells.
381
    inner: codec::ChannelCodec,
382
    /// The digest of bytes sent on this channel.
383
    ///
384
    /// Will be used for the SLOG or CLOG of the AUTHENTICATE cell.
385
    send_log: Option<ll::d::Sha256>,
386
    /// The digest of bytes received on this channel.
387
    ///
388
    /// Will be used for the SLOG or CLOG of the AUTHENTICATE cell.
389
    recv_log: Option<ll::d::Sha256>,
390
}
391

            
392
impl HandshakeChannelHandler {
393
    /// Constructor
394
93
    fn new(new_handler: &mut NewChannelHandler, link_version: LinkVersion) -> Self {
395
93
        Self {
396
93
            filter: MessageFilter::new(
397
93
                link_version,
398
93
                new_handler.channel_type,
399
93
                super::msg::MessageStage::Handshake,
400
93
            ),
401
93
            send_log: new_handler.send_log.take(),
402
93
            recv_log: new_handler.recv_log.take(),
403
93
            inner: codec::ChannelCodec::new(link_version.value()),
404
93
        }
405
93
    }
406

            
407
    /// Internal helper: Take a SHA256 digest and finalize it if any. None is returned if no log
408
    /// digest is given.
409
16
    fn finalize_log(log: Option<ll::d::Sha256>) -> Option<[u8; 32]> {
410
16
        log.map(|sha256| sha256.finalize().into())
411
16
    }
412

            
413
    /// Return an open handshake handler.
414
63
    fn next_handler(&mut self) -> OpenChannelHandler {
415
63
        OpenChannelHandler::new(
416
63
            self.inner
417
63
                .link_version()
418
63
                .try_into()
419
63
                .expect("Channel Codec with unknown link version"),
420
63
            self.channel_type(),
421
        )
422
63
    }
423

            
424
    /// The digest of bytes sent on this channel.
425
    ///
426
    /// This should only ever be called once as it consumes the send log.
427
    ///
428
    /// This will return `None` if one of:
429
    /// - The channel is not recording the send log.
430
    /// - The send log digest has already been taken.
431
6
    pub(crate) fn take_send_log_digest(&mut self) -> Option<AuthLogDigest> {
432
6
        Self::finalize_log(self.send_log.take())
433
6
    }
434

            
435
    /// The digest of bytes received on this channel.
436
    ///
437
    /// This should only ever be called once as it consumes the receive log.
438
    ///
439
    /// This will return `None` if one of:
440
    /// - The channel is not recording the receive log.
441
    /// - The receive log digest has already been taken.
442
10
    pub(crate) fn take_recv_log_digest(&mut self) -> Option<AuthLogDigest> {
443
10
        Self::finalize_log(self.recv_log.take())
444
10
    }
445

            
446
    /// Return the [`ChannelType`] of this handler.
447
63
    pub(crate) fn channel_type(&self) -> ChannelType {
448
63
        self.filter.channel_type()
449
63
    }
450

            
451
    /// Mark this handler as authenticated.
452
    pub(crate) fn set_authenticated(&mut self) {
453
        self.filter.channel_type_mut().set_authenticated();
454
    }
455
}
456

            
457
impl asynchronous_codec::Encoder for HandshakeChannelHandler {
458
    type Item<'a> = AnyChanCell;
459
    type Error = ChanError;
460

            
461
56
    fn encode(
462
56
        &mut self,
463
56
        item: Self::Item<'_>,
464
56
        dst: &mut BytesMut,
465
56
    ) -> std::result::Result<(), Self::Error> {
466
56
        let before_dst_len = dst.len();
467
56
        self.filter.encode_cell(item, &mut self.inner, dst)?;
468
56
        let after_dst_len = dst.len();
469
56
        if let Some(send_log) = self.send_log.as_mut() {
470
6
            // Only use what we actually wrote. Variable length cell are not padded and thus this
471
6
            // won't catch a bunch of padding.
472
6
            send_log.update(&dst[before_dst_len..after_dst_len]);
473
56
        }
474
56
        Ok(())
475
56
    }
476
}
477

            
478
impl asynchronous_codec::Decoder for HandshakeChannelHandler {
479
    type Item = AnyChanCell;
480
    type Error = ChanError;
481

            
482
860
    fn decode(
483
860
        &mut self,
484
860
        src: &mut BytesMut,
485
860
    ) -> std::result::Result<Option<Self::Item>, Self::Error> {
486
        // Remember the original src length so that we can later check if it changed.
487
860
        let original_len = src.len();
488

            
489
860
        let opt_cell = self.filter.decode_cell(&mut self.inner, src)?;
490

            
491
856
        let Some((cell, cell_bytes)) = opt_cell else {
492
            // If there was no cell, no bytes should have been removed from `src`.
493
            // If bytes were removed without us here knowing what they were,
494
            // then our `recv_log` won't be valid and our handshake will fail.
495
            // This is almost certainly a bug,
496
            // so let's fail early with an actionable error message.
497
660
            if self.recv_log.is_some() && src.len() != original_len {
498
                return Err(
499
                    internal!("Read bytes from the buffer without returning a cell").into(),
500
                );
501
660
            }
502
660
            return Ok(None);
503
        };
504

            
505
196
        if let Some(recv_log) = self.recv_log.as_mut() {
506
20
            recv_log.update(&cell_bytes);
507
176
        }
508

            
509
196
        Ok(Some(cell))
510
860
    }
511
}
512

            
513
/// The open channel handler which is used to decode and encode cells onto an open Channel.
514
pub(crate) struct OpenChannelHandler {
515
    /// Message filter used to allow or not a certain message.
516
    filter: MessageFilter,
517
    /// The cell codec that we'll use to encode and decode our cells.
518
    inner: codec::ChannelCodec,
519
}
520

            
521
impl OpenChannelHandler {
522
    /// Constructor
523
75
    fn new(link_version: LinkVersion, channel_type: ChannelType) -> Self {
524
75
        Self {
525
75
            inner: codec::ChannelCodec::new(link_version.value()),
526
75
            filter: MessageFilter::new(link_version, channel_type, super::msg::MessageStage::Open),
527
75
        }
528
75
    }
529

            
530
    /// Return the [`ChannelType`] of this handler.
531
51
    fn channel_type(&self) -> ChannelType {
532
51
        self.filter.channel_type()
533
51
    }
534
}
535

            
536
impl asynchronous_codec::Encoder for OpenChannelHandler {
537
    type Item<'a> = AnyChanCell;
538
    type Error = ChanError;
539

            
540
6
    fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
541
6
        self.filter.encode_cell(item, &mut self.inner, dst)
542
6
    }
543
}
544

            
545
impl asynchronous_codec::Decoder for OpenChannelHandler {
546
    type Item = AnyChanCell;
547
    type Error = ChanError;
548

            
549
12
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
550
12
        self.filter
551
12
            .decode_cell(&mut self.inner, src)
552
12
            .map(|x| x.map(|(msg, _msg_bytes)| msg))
553
12
    }
554
}
555

            
556
#[cfg(test)]
557
pub(crate) mod test {
558
    #![allow(clippy::unwrap_used)]
559
    use bytes::BytesMut;
560
    use digest::Digest;
561
    use futures::io::{AsyncRead, AsyncWrite, Cursor, Result};
562
    use futures::sink::SinkExt;
563
    use futures::stream::StreamExt;
564
    use futures::task::{Context, Poll};
565
    use hex_literal::hex;
566
    use std::pin::Pin;
567

            
568
    use tor_bytes::Writer;
569
    use tor_llcrypto as ll;
570
    use tor_rtcompat::StreamOps;
571

            
572
    use crate::channel::msg::LinkVersion;
573
    use crate::channel::{ChannelType, new_frame};
574

            
575
    use super::{ChannelCellHandler, OpenChannelHandler};
576
    use tor_cell::chancell::{AnyChanCell, ChanCmd, ChanMsg, CircId, msg};
577

            
578
    /// Helper type for reading and writing bytes to/from buffers.
579
    pub(crate) struct MsgBuf {
580
        /// Data we have received as a reader.
581
        inbuf: futures::io::Cursor<Vec<u8>>,
582
        /// Data we write as a writer.
583
        outbuf: futures::io::Cursor<Vec<u8>>,
584
    }
585

            
586
    impl AsyncRead for MsgBuf {
587
        fn poll_read(
588
            mut self: Pin<&mut Self>,
589
            cx: &mut Context<'_>,
590
            buf: &mut [u8],
591
        ) -> Poll<Result<usize>> {
592
            Pin::new(&mut self.inbuf).poll_read(cx, buf)
593
        }
594
    }
595
    impl AsyncWrite for MsgBuf {
596
        fn poll_write(
597
            mut self: Pin<&mut Self>,
598
            cx: &mut Context<'_>,
599
            buf: &[u8],
600
        ) -> Poll<Result<usize>> {
601
            Pin::new(&mut self.outbuf).poll_write(cx, buf)
602
        }
603
        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
604
            Pin::new(&mut self.outbuf).poll_flush(cx)
605
        }
606
        fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
607
            Pin::new(&mut self.outbuf).poll_close(cx)
608
        }
609
    }
610

            
611
    impl StreamOps for MsgBuf {}
612

            
613
    impl MsgBuf {
614
        pub(crate) fn new<T: Into<Vec<u8>>>(output: T) -> Self {
615
            let inbuf = Cursor::new(output.into());
616
            let outbuf = Cursor::new(Vec::new());
617
            MsgBuf { inbuf, outbuf }
618
        }
619

            
620
        pub(crate) fn consumed(&self) -> usize {
621
            self.inbuf.position() as usize
622
        }
623

            
624
        pub(crate) fn all_consumed(&self) -> bool {
625
            self.inbuf.get_ref().len() == self.consumed()
626
        }
627

            
628
        pub(crate) fn into_response(self) -> Vec<u8> {
629
            self.outbuf.into_inner()
630
        }
631
    }
632

            
633
    fn new_client_open_frame(
634
        mbuf: MsgBuf,
635
    ) -> asynchronous_codec::Framed<MsgBuf, ChannelCellHandler> {
636
        let open_handler = ChannelCellHandler::Open(OpenChannelHandler::new(
637
            LinkVersion::V5,
638
            ChannelType::ClientInitiator,
639
        ));
640
        asynchronous_codec::Framed::new(mbuf, open_handler)
641
    }
642

            
643
    #[test]
644
    fn check_client_encoding() {
645
        tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
646
            let mb = MsgBuf::new(&b""[..]);
647
            let mut framed = new_client_open_frame(mb);
648

            
649
            let destroycell = msg::Destroy::new(2.into());
650
            framed
651
                .send(AnyChanCell::new(CircId::new(7), destroycell.into()))
652
                .await
653
                .unwrap();
654

            
655
            framed.flush().await.unwrap();
656

            
657
            let data = framed.into_inner().into_response();
658

            
659
            assert_eq!(&data[0..10], &hex!("00000007 04 0200000000")[..]);
660
        });
661
    }
662

            
663
    #[test]
664
    fn check_client_decoding() {
665
        tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
666
            let mut dat = Vec::new();
667
            // DESTROY cell.
668
            dat.extend_from_slice(&hex!("00000007 04 0200000000")[..]);
669
            dat.resize(514, 0);
670
            let mb = MsgBuf::new(&dat[..]);
671
            let mut framed = new_client_open_frame(mb);
672

            
673
            let destroy = framed.next().await.unwrap().unwrap();
674

            
675
            let circ_id = CircId::new(7);
676
            assert_eq!(destroy.circid(), circ_id);
677
            assert_eq!(destroy.msg().cmd(), ChanCmd::DESTROY);
678

            
679
            assert!(framed.into_inner().all_consumed());
680
        });
681
    }
682

            
683
    #[test]
684
    fn handler_transition() {
685
        // Start as a client initiating a channel to a relay.
686
        let mut handler: ChannelCellHandler = ChannelType::ClientInitiator.into();
687
        assert!(matches!(handler, ChannelCellHandler::New(_)));
688

            
689
        // Set the link version protocol. Should transition to Handshake.
690
        let r = handler.set_link_version(5);
691
        assert!(r.is_ok());
692
        assert!(matches!(handler, ChannelCellHandler::Handshake(_)));
693

            
694
        // Set the link version protocol.
695
        let r = handler.set_open();
696
        assert!(r.is_ok());
697
        assert!(matches!(handler, ChannelCellHandler::Open(_)));
698
    }
699

            
700
    #[test]
701
    fn clog_digest() {
702
        tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
703
            let mut our_clog = ll::d::Sha256::new();
704
            let mbuf = MsgBuf::new(*b"");
705
            let mut frame = new_frame(mbuf, ChannelType::RelayInitiator);
706

            
707
            // This is a VERSIONS cell with value 5 in it.
708
            our_clog.update(hex!("0000 07 0002 0005"));
709
            let version_cell = AnyChanCell::new(
710
                None,
711
                msg::Versions::new(vec![5]).expect("Fail VERSIONS").into(),
712
            );
713
            let _ = frame.send(version_cell).await.unwrap();
714

            
715
            frame
716
                .codec_mut()
717
                .set_link_version(5)
718
                .expect("Fail link version set");
719

            
720
            // This is what an empty CERTS cell looks like.
721
            our_clog.update(hex!("0000 0000 81 0001 00"));
722
            let certs_cell = msg::Certs::new_empty();
723
            frame
724
                .send(AnyChanCell::new(None, certs_cell.into()))
725
                .await
726
                .unwrap();
727

            
728
            // Final CLOG should match.
729
            let clog_hash: [u8; 32] = our_clog.finalize().into();
730
            assert_eq!(frame.codec_mut().take_send_log_digest().unwrap(), clog_hash);
731
        });
732
    }
733

            
734
    #[test]
735
    fn slog_digest() {
736
        tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
737
            let mut our_slog = ll::d::Sha256::new();
738

            
739
            // Build a VERSIONS cell to start with.
740
            let mut data = BytesMut::new();
741
            data.extend_from_slice(
742
                msg::Versions::new(vec![5])
743
                    .unwrap()
744
                    .encode_for_handshake()
745
                    .expect("Fail VERSIONS encoding")
746
                    .as_slice(),
747
            );
748
            our_slog.update(&data);
749

            
750
            let mbuf = MsgBuf::new(data);
751
            let mut frame = new_frame(mbuf, ChannelType::RelayInitiator);
752

            
753
            // Receive the VERSIONS
754
            let _ = frame.next().await.transpose().expect("Fail to get cell");
755
            // Set the link version which will move the handler to Handshake state and then we'll be
756
            // able to decode the AUTH_CHALLENGE.
757
            frame
758
                .codec_mut()
759
                .set_link_version(5)
760
                .expect("Fail link version set");
761

            
762
            // Setup a new buffer for the next cell.
763
            let mut data = BytesMut::new();
764
            // This is a variable length cell with a wide circ ID of 0.
765
            data.write_u32(0);
766
            data.write_u8(ChanCmd::AUTH_CHALLENGE.into());
767
            data.write_u16(36); // This is the length of the payload.
768
            msg::AuthChallenge::new([42_u8; 32], vec![3])
769
                .encode_onto(&mut data)
770
                .expect("Fail AUTH_CHALLENGE encoding");
771
            our_slog.update(&data);
772

            
773
            // Change the I/O part of the Framed with this new buffer containing our new cell.
774
            *frame = MsgBuf::new(data);
775
            // Receive the AUTH_CHALLENGE
776
            let _ = frame.next().await.transpose().expect("Fail to get cell");
777

            
778
            // Final SLOG should match.
779
            let slog_hash: [u8; 32] = our_slog.finalize().into();
780
            assert_eq!(frame.codec_mut().take_recv_log_digest().unwrap(), slog_hash);
781
        });
782
    }
783
}