1use digest::Digest;
5use tor_bytes::Reader;
6use tor_cell::chancell::{
7 AnyChanCell, ChanCell, ChanCmd, ChanMsg, codec,
8 msg::{self, AnyChanMsg},
9};
10use tor_error::internal;
11use tor_llcrypto as ll;
12
13use bytes::BytesMut;
14
15use crate::{channel::msg::LinkVersion, util::err::Error as ChanError};
16
17use super::{ChannelType, msg::MessageFilter};
18
19pub(crate) type AuthLogDigest = [u8; 32];
21#[derive(Debug, PartialEq)]
23pub(crate) struct ClogDigest(AuthLogDigest);
24#[derive(Debug, PartialEq)]
26pub(crate) struct SlogDigest(AuthLogDigest);
27
28impl ClogDigest {
29 pub(crate) fn new(digest: AuthLogDigest) -> Self {
31 Self(digest)
32 }
33}
34
35impl SlogDigest {
36 pub(crate) fn new(digest: AuthLogDigest) -> Self {
38 Self(digest)
39 }
40}
41
42impl AsRef<[u8]> for ClogDigest {
43 fn as_ref(&self) -> &[u8] {
44 &self.0
45 }
46}
47impl AsRef<[u8]> for SlogDigest {
48 fn as_ref(&self) -> &[u8] {
49 &self.0
50 }
51}
52
53pub(crate) enum ChannelCellHandler {
61 New(NewChannelHandler),
64 Handshake(HandshakeChannelHandler),
67 Open(OpenChannelHandler),
69}
70
71impl From<super::ChannelType> for ChannelCellHandler {
74 fn from(ty: ChannelType) -> Self {
75 Self::New(ty.into())
76 }
77}
78
79impl ChannelCellHandler {
80 pub(crate) fn channel_type(&self) -> ChannelType {
82 match self {
83 Self::New(h) => h.channel_type,
84 Self::Handshake(h) => h.channel_type(),
85 Self::Open(h) => h.channel_type(),
86 }
87 }
88
89 pub(crate) fn set_link_version(&mut self, link_version: u16) -> Result<(), ChanError> {
95 let Self::New(new_handler) = self else {
96 return Err(ChanError::Bug(internal!(
97 "Setting link protocol without a new handler",
98 )));
99 };
100 *self = Self::Handshake(new_handler.next_handler(link_version.try_into()?));
101 Ok(())
102 }
103
104 pub(crate) fn set_open(&mut self) -> Result<(), ChanError> {
108 let Self::Handshake(handler) = self else {
109 return Err(ChanError::Bug(internal!(
110 "Setting open without a handshake handler"
111 )));
112 };
113 *self = Self::Open(handler.next_handler());
114 Ok(())
115 }
116
117 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 pub(crate) fn take_send_log_digest(&mut self) -> Result<AuthLogDigest, ChanError> {
140 if let Self::Handshake(handler) = self {
141 handler
142 .take_send_log_digest()
143 .ok_or(ChanError::Bug(internal!(
144 "No send log digest on channel, or already taken"
145 )))
146 } else {
147 Err(ChanError::Bug(internal!(
148 "Getting send log digest without a handshake handler"
149 )))
150 }
151 }
152
153 pub(crate) fn take_recv_log_digest(&mut self) -> Result<AuthLogDigest, ChanError> {
162 if let Self::Handshake(handler) = self {
163 handler
164 .take_recv_log_digest()
165 .ok_or(ChanError::Bug(internal!(
166 "No recv log digest on channel, or already taken"
167 )))
168 } else {
169 Err(ChanError::Bug(internal!(
170 "Getting recv log digest without a handshake handler"
171 )))
172 }
173 }
174}
175
176impl asynchronous_codec::Decoder for ChannelCellHandler {
195 type Item = AnyChanCell;
196 type Error = ChanError;
197
198 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
199 match self {
200 Self::New(c) => c
201 .decode(src)
202 .map(|opt| opt.map(|msg| ChanCell::new(None, msg.into()))),
203 Self::Handshake(c) => c.decode(src),
204 Self::Open(c) => c.decode(src),
205 }
206 }
207}
208
209impl asynchronous_codec::Encoder for ChannelCellHandler {
210 type Item<'a> = AnyChanCell;
211 type Error = ChanError;
212
213 fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
214 match self {
215 Self::New(c) => {
216 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 c.encode(versions, dst)
224 }
225 Self::Handshake(c) => c.encode(item, dst),
226 Self::Open(c) => c.encode(item, dst),
227 }
228 }
229}
230
231pub(crate) struct NewChannelHandler {
236 channel_type: ChannelType,
238 send_log: Option<ll::d::Sha256>,
242 recv_log: Option<ll::d::Sha256>,
246}
247
248impl NewChannelHandler {
249 fn next_handler(&mut self, link_version: LinkVersion) -> HandshakeChannelHandler {
251 HandshakeChannelHandler::new(self, link_version)
252 }
253}
254
255impl From<ChannelType> for NewChannelHandler {
256 fn from(channel_type: ChannelType) -> Self {
257 match channel_type {
258 ChannelType::ClientInitiator => Self {
259 channel_type,
260 send_log: None,
261 recv_log: None,
262 },
263 ChannelType::RelayInitiator | ChannelType::RelayResponder { .. } => Self {
266 channel_type,
267 send_log: Some(ll::d::Sha256::new()),
268 recv_log: Some(ll::d::Sha256::new()),
269 },
270 }
271 }
272}
273
274impl asynchronous_codec::Decoder for NewChannelHandler {
275 type Item = msg::Versions;
276 type Error = ChanError;
277
278 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
279 const HEADER_SIZE: usize = 5;
287
288 if src.len() < HEADER_SIZE {
292 return Ok(None);
293 }
294
295 let circ_id = u16::from_be_bytes([src[0], src[1]]);
298 if circ_id != 0 {
299 return Err(Self::Error::HandshakeProto(
300 "Invalid CircID in variable cell".into(),
301 ));
302 }
303
304 let cmd = ChanCmd::from(src[2]);
307 if cmd != ChanCmd::VERSIONS {
308 return Err(Self::Error::HandshakeProto(format!(
309 "Invalid command {cmd} variable cell, expected a VERSIONS."
310 )));
311 }
312
313 let body_len = u16::from_be_bytes([src[3], src[4]]) as usize;
316
317 if body_len % 2 == 1 {
322 return Err(Self::Error::HandshakeProto(
323 "VERSIONS cell body length is odd. Rejecting.".into(),
324 ));
325 }
326
327 let wanted_bytes = HEADER_SIZE + body_len;
329 if src.len() < wanted_bytes {
330 return Ok(None);
335 }
336 let mut data = src.split_to(wanted_bytes);
338
339 if let Some(recv_log) = self.recv_log.as_mut() {
343 recv_log.update(&data);
344 }
345
346 let body = data.split_off(HEADER_SIZE).freeze();
348 let mut reader = Reader::from_bytes(&body);
349
350 let cell = msg::Versions::decode_from_reader(cmd, &mut reader)
352 .map_err(|e| Self::Error::from_bytes_err(e, "new cell handler"))?;
353 Ok(Some(cell))
354 }
355}
356
357impl asynchronous_codec::Encoder for NewChannelHandler {
358 type Item<'a> = msg::Versions;
359 type Error = ChanError;
360
361 fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
362 let encoded_bytes = item
363 .encode_for_handshake()
364 .map_err(|e| Self::Error::from_bytes_enc(e, "new cell handler"))?;
365 if let Some(send_log) = self.send_log.as_mut() {
367 send_log.update(&encoded_bytes);
368 }
369 dst.extend_from_slice(&encoded_bytes);
371 Ok(())
372 }
373}
374
375pub(crate) struct HandshakeChannelHandler {
378 filter: MessageFilter,
380 inner: codec::ChannelCodec,
382 send_log: Option<ll::d::Sha256>,
386 recv_log: Option<ll::d::Sha256>,
390}
391
392impl HandshakeChannelHandler {
393 fn new(new_handler: &mut NewChannelHandler, link_version: LinkVersion) -> Self {
395 Self {
396 filter: MessageFilter::new(
397 link_version,
398 new_handler.channel_type,
399 super::msg::MessageStage::Handshake,
400 ),
401 send_log: new_handler.send_log.take(),
402 recv_log: new_handler.recv_log.take(),
403 inner: codec::ChannelCodec::new(link_version.value()),
404 }
405 }
406
407 fn finalize_log(log: Option<ll::d::Sha256>) -> Option<[u8; 32]> {
410 log.map(|sha256| sha256.finalize().into())
411 }
412
413 fn next_handler(&mut self) -> OpenChannelHandler {
415 OpenChannelHandler::new(
416 self.inner
417 .link_version()
418 .try_into()
419 .expect("Channel Codec with unknown link version"),
420 self.channel_type(),
421 )
422 }
423
424 pub(crate) fn take_send_log_digest(&mut self) -> Option<AuthLogDigest> {
432 Self::finalize_log(self.send_log.take())
433 }
434
435 pub(crate) fn take_recv_log_digest(&mut self) -> Option<AuthLogDigest> {
443 Self::finalize_log(self.recv_log.take())
444 }
445
446 pub(crate) fn channel_type(&self) -> ChannelType {
448 self.filter.channel_type()
449 }
450
451 pub(crate) fn set_authenticated(&mut self) {
453 self.filter.channel_type_mut().set_authenticated();
454 }
455}
456
457impl asynchronous_codec::Encoder for HandshakeChannelHandler {
458 type Item<'a> = AnyChanCell;
459 type Error = ChanError;
460
461 fn encode(
462 &mut self,
463 item: Self::Item<'_>,
464 dst: &mut BytesMut,
465 ) -> std::result::Result<(), Self::Error> {
466 let before_dst_len = dst.len();
467 self.filter.encode_cell(item, &mut self.inner, dst)?;
468 let after_dst_len = dst.len();
469 if let Some(send_log) = self.send_log.as_mut() {
470 send_log.update(&dst[before_dst_len..after_dst_len]);
473 }
474 Ok(())
475 }
476}
477
478impl asynchronous_codec::Decoder for HandshakeChannelHandler {
479 type Item = AnyChanCell;
480 type Error = ChanError;
481
482 fn decode(
483 &mut self,
484 src: &mut BytesMut,
485 ) -> std::result::Result<Option<Self::Item>, Self::Error> {
486 let orig = src.clone(); let cell = self.filter.decode_cell(&mut self.inner, src)?;
488 if let Some(recv_log) = self.recv_log.as_mut() {
489 let n_used = orig.len() - src.len();
490 recv_log.update(&orig[..n_used]);
491 }
492 Ok(cell)
493 }
494}
495
496pub(crate) struct OpenChannelHandler {
498 filter: MessageFilter,
500 inner: codec::ChannelCodec,
502}
503
504impl OpenChannelHandler {
505 fn new(link_version: LinkVersion, channel_type: ChannelType) -> Self {
507 Self {
508 inner: codec::ChannelCodec::new(link_version.value()),
509 filter: MessageFilter::new(link_version, channel_type, super::msg::MessageStage::Open),
510 }
511 }
512
513 fn channel_type(&self) -> ChannelType {
515 self.filter.channel_type()
516 }
517}
518
519impl asynchronous_codec::Encoder for OpenChannelHandler {
520 type Item<'a> = AnyChanCell;
521 type Error = ChanError;
522
523 fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
524 self.filter.encode_cell(item, &mut self.inner, dst)
525 }
526}
527
528impl asynchronous_codec::Decoder for OpenChannelHandler {
529 type Item = AnyChanCell;
530 type Error = ChanError;
531
532 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
533 self.filter.decode_cell(&mut self.inner, src)
534 }
535}
536
537#[cfg(test)]
538pub(crate) mod test {
539 #![allow(clippy::unwrap_used)]
540 use bytes::BytesMut;
541 use digest::Digest;
542 use futures::io::{AsyncRead, AsyncWrite, Cursor, Result};
543 use futures::sink::SinkExt;
544 use futures::stream::StreamExt;
545 use futures::task::{Context, Poll};
546 use hex_literal::hex;
547 use std::pin::Pin;
548
549 use tor_bytes::Writer;
550 use tor_llcrypto as ll;
551 use tor_rtcompat::StreamOps;
552
553 use crate::channel::msg::LinkVersion;
554 use crate::channel::{ChannelType, new_frame};
555
556 use super::{ChannelCellHandler, OpenChannelHandler};
557 use tor_cell::chancell::{AnyChanCell, ChanCmd, ChanMsg, CircId, msg};
558
559 pub(crate) struct MsgBuf {
561 inbuf: futures::io::Cursor<Vec<u8>>,
563 outbuf: futures::io::Cursor<Vec<u8>>,
565 }
566
567 impl AsyncRead for MsgBuf {
568 fn poll_read(
569 mut self: Pin<&mut Self>,
570 cx: &mut Context<'_>,
571 buf: &mut [u8],
572 ) -> Poll<Result<usize>> {
573 Pin::new(&mut self.inbuf).poll_read(cx, buf)
574 }
575 }
576 impl AsyncWrite for MsgBuf {
577 fn poll_write(
578 mut self: Pin<&mut Self>,
579 cx: &mut Context<'_>,
580 buf: &[u8],
581 ) -> Poll<Result<usize>> {
582 Pin::new(&mut self.outbuf).poll_write(cx, buf)
583 }
584 fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
585 Pin::new(&mut self.outbuf).poll_flush(cx)
586 }
587 fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
588 Pin::new(&mut self.outbuf).poll_close(cx)
589 }
590 }
591
592 impl StreamOps for MsgBuf {}
593
594 impl MsgBuf {
595 pub(crate) fn new<T: Into<Vec<u8>>>(output: T) -> Self {
596 let inbuf = Cursor::new(output.into());
597 let outbuf = Cursor::new(Vec::new());
598 MsgBuf { inbuf, outbuf }
599 }
600
601 pub(crate) fn consumed(&self) -> usize {
602 self.inbuf.position() as usize
603 }
604
605 pub(crate) fn all_consumed(&self) -> bool {
606 self.inbuf.get_ref().len() == self.consumed()
607 }
608
609 pub(crate) fn into_response(self) -> Vec<u8> {
610 self.outbuf.into_inner()
611 }
612 }
613
614 fn new_client_open_frame(
615 mbuf: MsgBuf,
616 ) -> asynchronous_codec::Framed<MsgBuf, ChannelCellHandler> {
617 let open_handler = ChannelCellHandler::Open(OpenChannelHandler::new(
618 LinkVersion::V5,
619 ChannelType::ClientInitiator,
620 ));
621 asynchronous_codec::Framed::new(mbuf, open_handler)
622 }
623
624 #[test]
625 fn check_client_encoding() {
626 tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
627 let mb = MsgBuf::new(&b""[..]);
628 let mut framed = new_client_open_frame(mb);
629
630 let destroycell = msg::Destroy::new(2.into());
631 framed
632 .send(AnyChanCell::new(CircId::new(7), destroycell.into()))
633 .await
634 .unwrap();
635
636 framed.flush().await.unwrap();
637
638 let data = framed.into_inner().into_response();
639
640 assert_eq!(&data[0..10], &hex!("00000007 04 0200000000")[..]);
641 });
642 }
643
644 #[test]
645 fn check_client_decoding() {
646 tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
647 let mut dat = Vec::new();
648 dat.extend_from_slice(&hex!("00000007 04 0200000000")[..]);
650 dat.resize(514, 0);
651 let mb = MsgBuf::new(&dat[..]);
652 let mut framed = new_client_open_frame(mb);
653
654 let destroy = framed.next().await.unwrap().unwrap();
655
656 let circ_id = CircId::new(7);
657 assert_eq!(destroy.circid(), circ_id);
658 assert_eq!(destroy.msg().cmd(), ChanCmd::DESTROY);
659
660 assert!(framed.into_inner().all_consumed());
661 });
662 }
663
664 #[test]
665 fn handler_transition() {
666 let mut handler: ChannelCellHandler = ChannelType::ClientInitiator.into();
668 assert!(matches!(handler, ChannelCellHandler::New(_)));
669
670 let r = handler.set_link_version(5);
672 assert!(r.is_ok());
673 assert!(matches!(handler, ChannelCellHandler::Handshake(_)));
674
675 let r = handler.set_open();
677 assert!(r.is_ok());
678 assert!(matches!(handler, ChannelCellHandler::Open(_)));
679 }
680
681 #[test]
682 fn clog_digest() {
683 tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
684 let mut our_clog = ll::d::Sha256::new();
685 let mbuf = MsgBuf::new(*b"");
686 let mut frame = new_frame(mbuf, ChannelType::RelayInitiator);
687
688 our_clog.update(hex!("0000 07 0002 0005"));
690 let version_cell = AnyChanCell::new(
691 None,
692 msg::Versions::new(vec![5]).expect("Fail VERSIONS").into(),
693 );
694 let _ = frame.send(version_cell).await.unwrap();
695
696 frame
697 .codec_mut()
698 .set_link_version(5)
699 .expect("Fail link version set");
700
701 our_clog.update(hex!("0000 0000 81 0001 00"));
703 let certs_cell = msg::Certs::new_empty();
704 frame
705 .send(AnyChanCell::new(None, certs_cell.into()))
706 .await
707 .unwrap();
708
709 let clog_hash: [u8; 32] = our_clog.finalize().into();
711 assert_eq!(frame.codec_mut().take_send_log_digest().unwrap(), clog_hash);
712 });
713 }
714
715 #[test]
716 fn slog_digest() {
717 tor_rtcompat::test_with_all_runtimes!(|_rt| async move {
718 let mut our_slog = ll::d::Sha256::new();
719
720 let mut data = BytesMut::new();
722 data.extend_from_slice(
723 msg::Versions::new(vec![5])
724 .unwrap()
725 .encode_for_handshake()
726 .expect("Fail VERSIONS encoding")
727 .as_slice(),
728 );
729 our_slog.update(&data);
730
731 let mbuf = MsgBuf::new(data);
732 let mut frame = new_frame(mbuf, ChannelType::RelayInitiator);
733
734 let _ = frame.next().await.transpose().expect("Fail to get cell");
736 frame
739 .codec_mut()
740 .set_link_version(5)
741 .expect("Fail link version set");
742
743 let mut data = BytesMut::new();
745 data.write_u32(0);
747 data.write_u8(ChanCmd::AUTH_CHALLENGE.into());
748 data.write_u16(36); msg::AuthChallenge::new([42_u8; 32], vec![3])
750 .encode_onto(&mut data)
751 .expect("Fail AUTH_CHALLENGE encoding");
752 our_slog.update(&data);
753
754 *frame = MsgBuf::new(data);
756 let _ = frame.next().await.transpose().expect("Fail to get cell");
758
759 let slog_hash: [u8; 32] = our_slog.finalize().into();
761 assert_eq!(frame.codec_mut().take_recv_log_digest().unwrap(), slog_hash);
762 });
763 }
764}