WIP: extend.

tor-proto/src/circuit.rs

@@ -8,9 +8,11 @@ use crate::chancell::{
 use crate::channel::Channel;
 use crate::crypto::cell::{ClientLayer, CryptInit};
 use crate::crypto::handshake::{ClientHandshake, KeyGenerator};
-use crate::relaycell::msg::RelayCell;
+use crate::relaycell::{msg::RelayCell, msg::RelayMsg, StreamCmd};
 use crate::{Error, Result};
 
+use tor_linkspec::LinkSpec;
+
 use futures::channel::mpsc;
 use futures::io::{AsyncRead, AsyncWrite};
 use futures::stream::StreamExt;
@@ -151,6 +153,74 @@ where
         Ok(())
     }
 
+    /// Helper: extend the circuit.
+    async fn extend_impl<R, L, H>(
+        &mut self,
+        rng: &mut R,
+        handshake_id: u16,
+        key: &H::KeyType,
+        linkspecs: Vec<LinkSpec>,
+    ) -> Result<()>
+    where
+        R: Rng + CryptoRng,
+        L: CryptInit + ClientLayer + 'static,
+        H: ClientHandshake,
+        H::KeyGen: KeyGenerator,
+    {
+        use crate::relaycell::msg::{Body, Extend2};
+        if self.crypto.n_layers() == 0 {
+            return Err(Error::CircExtend("Circuit not yet created"));
+        }
+        let hop = (self.crypto.n_layers() - 1) as u8;
+
+        let (state, msg) = H::client1(rng, &key)?;
+        // XXXX sort the linkspecs into canonical order
+        let extend_msg = Extend2::new(linkspecs, handshake_id, msg);
+        let cell = RelayCell::new(0.into(), extend_msg.as_message());
+
+        // Send the message to the last hop...
+        self.send_relay_cell(
+            hop, true, // early
+            cell,
+        )
+        .await?;
+
+        // and wait for a response.
+        // XXXX This is no good for production use.  We shouldn't wait
+        // XXXX for the _NEXT_ relay cell, but instead for the next
+        // XXXX EXTENDED/EXTENDED2 cell.  Other relay cells should go
+        // XXXX elsewhere.
+        let (from_hop, cell) = self.recv_relay_cell().await?;
+
+        // Did we get the right response?
+        if from_hop != hop || cell.get_cmd() != StreamCmd::EXTENDED2 {
+            return Err(Error::CircProto(format!(
+                "wanted EXTENDED2 from {}; got {} from {}",
+                hop,
+                cell.get_cmd(),
+                from_hop
+            )));
+        }
+        let (streamid, msg) = cell.into_streamid_and_msg();
+        if streamid != 0.into() {
+            return Err(Error::CircProto(format!(
+                "got nonzero stream ID {} on EXTENDED2",
+                streamid
+            )));
+        }
+        let msg = match msg {
+            RelayMsg::Extended2(e) => e,
+            _ => return Err(Error::InternalError("body didn't match cmd".into())),
+        };
+        let server_handshake = msg.into_body();
+
+        let keygen = H::client2(state, server_handshake)?;
+        let layer = L::construct(keygen)?;
+
+        self.crypto.add_layer(Box::new(layer));
+        Ok(())
+    }
+
     /// Use the (questionable!) CREATE_FAST handshake to connect to the
     /// first hop of this circuit.
     ///

tor-proto/src/relaycell.rs

@@ -6,6 +6,7 @@ pub mod msg;
 
 caret_int! {
     /// A command that identifies the type of a relay cell
+    // XXXX maybe rename to CircCmd.
     pub struct StreamCmd(u8) {
         /// Start a new stream
         BEGIN = 1,
@@ -76,3 +77,9 @@ impl Into<u16> for StreamID {
         self.0
     }
 }
+
+impl std::fmt::Display for StreamID {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
+        self.0.fmt(f)
+    }
+}

tor-proto/src/relaycell/msg.rs

@@ -20,10 +20,25 @@ use rand::{CryptoRng, Rng};
 /// A parsed relay cell.
 pub struct RelayCell {
     streamid: StreamID,
-    body: RelayMsg,
+    body: RelayMsg, // XXX rename to msg.
 }
 
 impl RelayCell {
+    /// Construct a new relay cell.
+    pub fn new(streamid: StreamID, msg: RelayMsg) -> Self {
+        RelayCell {
+            streamid,
+            body: msg,
+        }
+    }
+    /// Consume this cell and return its components.
+    pub fn into_streamid_and_msg(self) -> (StreamID, RelayMsg) {
+        (self.streamid, self.body)
+    }
+    /// Return the command for this cell.
+    pub fn get_cmd(&self) -> StreamCmd {
+        self.body.get_cmd()
+    }
     /// Consume this relay message and encode it as a 509-byte padded cell
     /// body.
     pub fn encode<R: Rng + CryptoRng>(self, rng: &mut R) -> crate::Result<RelayCellBody> {
@@ -135,12 +150,12 @@ pub enum RelayMsg {
 }
 
 /// Internal: traits in common different cell bodies.
-trait Body: Sized {
+pub trait Body: Sized {
+    /// Convert this type into a RelayMsg, wrapped appropriate.
+    fn as_message(self) -> RelayMsg;
+    /// Decode a relay cell body from a provided reader.
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self>;
-    fn decode(body: Vec<u8>) -> Result<Self> {
-        let mut reader = Reader::from_slice(&body[..]);
-        Self::decode_from_reader(&mut reader)
-    }
+    /// Encode the body of this cell into the end of a vec.
     fn encode_onto(self, w: &mut Vec<u8>);
 }
 
@@ -221,6 +236,9 @@ pub struct Begin {
 }
 
 impl Body for Begin {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Begin(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let addr = r.take_until(b':')?;
         let port = r.take_until(0)?;
@@ -257,14 +275,14 @@ pub struct Data {
 }
 
 impl Body for Data {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Data(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         Ok(Data {
             body: r.take(r.remaining())?.into(),
         })
     }
-    fn decode(body: Vec<u8>) -> Result<Self> {
-        Ok(Data { body })
-    }
     fn encode_onto(mut self, w: &mut Vec<u8>) {
         w.append(&mut self.body);
     }
@@ -278,6 +296,9 @@ pub struct End {
 const REASON_MISC: u8 = 1;
 const REASON_EXITPOLICY: u8 = 4;
 impl Body for End {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::End(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         if r.remaining() == 0 {
             return Ok(End {
@@ -322,6 +343,9 @@ pub struct Connected {
     addr: Option<(IpAddr, u32)>,
 }
 impl Body for Connected {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Connected(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         if r.remaining() == 0 {
             return Ok(Connected { addr: None });
@@ -362,6 +386,9 @@ pub struct Sendme {
 }
 
 impl Body for Sendme {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Sendme(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         Ok(Sendme {
             digest: Some(r.take(r.remaining())?.into()),
@@ -384,6 +411,9 @@ pub struct Extend {
 }
 
 impl Body for Extend {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Extend(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let addr = r.extract()?;
         let port = r.take_u16()?;
@@ -410,6 +440,9 @@ pub struct Extended {
 }
 
 impl Body for Extended {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Extended(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let handshake = r.take(TAP_S_HANDSHAKE_LEN)?.into();
         Ok(Extended { handshake })
@@ -421,28 +454,41 @@ impl Body for Extended {
 
 /// Extend the circuit to a new hop
 pub struct Extend2 {
-    ls: Vec<LinkSpec>,
+    linkspec: Vec<LinkSpec>,
     handshake_type: u16,
     handshake: Vec<u8>,
 }
+impl Extend2 {
+    /// Create a new Extend2 cell.
+    pub fn new(linkspec: Vec<LinkSpec>, handshake_type: u16, handshake: Vec<u8>) -> Self {
+        Extend2 {
+            linkspec,
+            handshake_type,
+            handshake,
+        }
+    }
+}
 
 impl Body for Extend2 {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Extend2(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let n = r.take_u8()?;
-        let ls = r.extract_n(n as usize)?;
+        let linkspec = r.extract_n(n as usize)?;
         let handshake_type = r.take_u16()?;
         let hlen = r.take_u16()?;
         let handshake = r.take(hlen as usize)?.into();
         Ok(Extend2 {
-            ls,
+            linkspec,
             handshake_type,
             handshake,
         })
     }
     fn encode_onto(self, w: &mut Vec<u8>) {
-        assert!(self.ls.len() <= std::u8::MAX as usize);
-        w.write_u8(self.ls.len() as u8);
-        for ls in self.ls.iter() {
+        assert!(self.linkspec.len() <= std::u8::MAX as usize);
+        w.write_u8(self.linkspec.len() as u8);
+        for ls in self.linkspec.iter() {
             w.write(ls);
         }
         w.write_u16(self.handshake_type);
@@ -454,8 +500,16 @@ impl Body for Extend2 {
 pub struct Extended2 {
     handshake: Vec<u8>,
 }
-
+impl Extended2 {
+    /// Consume this extended2 cell and return its body.
+    pub fn into_body(self) -> Vec<u8> {
+        self.handshake
+    }
+}
 impl Body for Extended2 {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Extended2(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let hlen = r.take_u16()?;
         let handshake = r.take(hlen as usize)?;
@@ -474,6 +528,9 @@ impl Body for Extended2 {
 pub struct Truncate {}
 
 impl Body for Truncate {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Truncate(self)
+    }
     fn decode_from_reader(_r: &mut Reader<'_>) -> Result<Self> {
         Ok(Truncate {})
     }
@@ -486,6 +543,9 @@ pub struct Truncated {
 }
 
 impl Body for Truncated {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Truncated(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         Ok(Truncated {
             reason: r.take_u8()?,
@@ -502,6 +562,9 @@ pub struct Resolve {
 }
 
 impl Body for Resolve {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Resolve(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let query = r.take_until(0)?;
         Ok(Resolve {
@@ -616,6 +679,9 @@ pub struct Resolved {
 }
 
 impl Body for Resolved {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Resolved(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         let mut answers = Vec::new();
         while r.remaining() > 0 {
@@ -653,6 +719,9 @@ impl Unrecognized {
 }
 
 impl Body for Unrecognized {
+    fn as_message(self) -> RelayMsg {
+        RelayMsg::Unrecognized(self)
+    }
     fn decode_from_reader(r: &mut Reader<'_>) -> Result<Self> {
         Ok(Unrecognized {
             cmd: 0.into(),

tor-proto/src/util/err.rs

@@ -38,6 +38,9 @@ pub enum Error {
     /// Protocol violation at the channel level
     #[error("channel protocol violation: {0}")]
     ChanProto(String),
+    /// Protocol violation at the circuit level
+    #[error("circuit protocol violation: {0}")]
+    CircProto(String),
     /// Circuit is closed.
     #[error("circuit closed")]
     CircuitClosed,