Work on client channel handshake: mostly done, except for verification

client-demo/src/main.rs

@@ -1,20 +1,18 @@
 mod err;
 
-use err::{Error, Result};
-
 use log::{info, LevelFilter};
 use std::path::PathBuf;
 use tor_linkspec::ChanTarget;
+use tor_proto::channel::{Channel, OutboundClientHandshake};
 
 //use async_std::prelude::*;
-use async_native_tls::TlsConnector;
+use async_native_tls::{TlsConnector, TlsStream};
 use async_std::net;
+use err::{Error, Result};
 
 use rand::thread_rng;
 
-pub struct Channel {}
-
-async fn connect<C: ChanTarget>(target: &C) -> Result<Channel> {
+async fn connect<C: ChanTarget>(target: &C) -> Result<Channel<TlsStream<net::TcpStream>>> {
     let addr = target
         .get_addrs()
         .get(0)
@@ -28,10 +26,17 @@ async fn connect<C: ChanTarget>(target: &C) -> Result<Channel> {
     let stream = net::TcpStream::connect(addr).await?;
 
     info!("Negotiating TLS with {}", addr);
-    let _tlscon = connector.connect("ignored", stream).await?;
+    let tlscon = connector.connect("ignored", stream).await?;
     info!("TLS negotiated.");
 
-    Ok(Channel {})
+    let chan = OutboundClientHandshake::new(tlscon).connect().await?;
+    info!("version negotiated and cells read.");
+    let chan = chan.check(target)?;
+    info!("Certs validated (only not really)");
+    let chan = chan.finish(&addr.ip()).await?;
+    info!("Channel complete.");
+
+    Ok(chan)
 }
 
 fn get_netdir() -> Result<tor_netdir::NetDir> {
@@ -52,13 +57,15 @@ fn main() -> Result<()> {
     simple_logging::log_to_stderr(LevelFilter::Info);
 
     let dir = get_netdir()?;
+    // TODO CONFORMANCE: we should stop now if there are required
+    // protovers we don't support.
 
     let g = dir
         .pick_relay(&mut thread_rng(), |_, u| u)
         .ok_or(Error::Misc("no usable relays"))?;
 
     async_std::task::block_on(async {
-        let _con = connect(&g).await?;
+        let _chan = connect(&g).await?;
 
         Ok(())
     })

tor-proto/Cargo.toml

@@ -29,6 +29,7 @@ sha2 = "0.9.1"
 futures_codec = "*"
 bytes = "*"
 thiserror = "*"
+futures = "*"
 
 [dev-dependencies]
 hex-literal = "0.3.1"

tor-proto/src/chancell.rs

@@ -35,6 +35,11 @@ impl Into<u32> for CircID {
         self.0
     }
 }
+impl std::fmt::Display for CircID {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
+        self.0.fmt(f)
+    }
+}
 
 caret_int! {
     /// A ChanCmd is the type of a channel cell.  The value of the ChanCmd
@@ -141,7 +146,7 @@ impl ChanCmd {
     /// circuit ID `id`.
     pub fn accepts_circid_val(self, id: CircID) -> bool {
         if self.is_recognized() {
-            self.allows_circid() == (id == 0.into())
+            self.allows_circid() != (id == 0.into())
         } else {
             true
         }
@@ -156,6 +161,10 @@ pub struct ChanCell {
 }
 
 impl ChanCell {
+    /// Construct a new channel cell.
+    pub fn new(circid: CircID, msg: msg::ChanMsg) -> Self {
+        ChanCell { circid, msg }
+    }
     /// Return the circuit ID for this cell.
     pub fn get_circid(&self) -> CircID {
         self.circid
@@ -164,4 +173,8 @@ impl ChanCell {
     pub fn get_msg(&self) -> &msg::ChanMsg {
         &self.msg
     }
+    /// Consume this cell and return its components.
+    pub fn into_circid_and_msg(self) -> (CircID, msg::ChanMsg) {
+        (self.circid, self.msg)
+    }
 }

tor-proto/src/chancell/codec.rs

@@ -1,10 +1,11 @@
 //! Implementation for asynchronus encoding and decoding of ChanCells.
 
-use crate::chancell::{msg::Body, ChanCell, ChanCmd, CircID};
+use crate::chancell::{msg, ChanCell, ChanCmd, CircID};
 use crate::crypto::cell::CELL_BODY_LEN;
 use crate::Error;
 use arrayref::{array_mut_ref, array_ref};
 use tor_bytes::{self, Reader, Writer};
+//use log::trace;
 
 /// This Codec object can be used with the `futures_codec` crate to
 /// turn an asynchronous byte stream into an asynchronous Stream/Sink pair
@@ -96,15 +97,17 @@ impl futures_codec::Decoder for ChannelCodec {
         }
 
         let cell = src.split_to(cell_len).freeze();
+        //trace!("{:?} cell body ({}) is {:?}", cmd, cell.len(), &cell[..]);
         let mut r = Reader::from_bytes(&cell);
         let circid: CircID = r.take_u32()?.into();
-        r.advance(if varcell { 1 } else { 3 })?;
-        let msg = r.extract()?;
+        r.advance(if varcell { 3 } else { 1 })?;
+        let msg = msg::ChanMsg::take(&mut r, cmd)?;
 
         if !cmd.accepts_circid_val(circid) {
-            return Err(Error::ChanProto(
-                "Invalid circuit ID for cell command".into(),
-            ));
+            return Err(Error::ChanProto(format!(
+                "Invalid circuit ID {} for cell command {}",
+                circid, cmd
+            )));
         }
         Ok(Some(ChanCell { circid, msg }))
     }

tor-proto/src/chancell/msg.rs

@@ -94,13 +94,9 @@ impl ChanMsg {
             Unrecognized(c) => c.get_cmd(),
         }
     }
-}
 
-impl Body for ChanMsg {
-    fn as_message(self) -> Self {
-        self
-    }
-    fn write_body_onto<W: Writer + ?Sized>(self, w: &mut W) {
+    /// Write the body of this message (not including length or command).
+    pub fn write_body_onto<W: Writer + ?Sized>(self, w: &mut W) {
         use ChanMsg::*;
         match self {
             Padding(b) => b.write_body_onto(w),
@@ -124,11 +120,11 @@ impl Body for ChanMsg {
             Unrecognized(b) => b.write_body_onto(w),
         }
     }
-}
 
-impl Readable for ChanMsg {
-    fn take_from(r: &mut Reader<'_>) -> Result<Self> {
-        let cmd = r.take_u8()?.into();
+    /// Decode this message from a given reader, according to a specified
+    /// command value. The reader must be truncated to the exact length
+    /// of the body.
+    pub fn take(r: &mut Reader<'_>, cmd: ChanCmd) -> Result<Self> {
         use ChanMsg::*;
         Ok(match cmd {
             ChanCmd::PADDING => Padding(r.extract()?),
@@ -425,6 +421,16 @@ fn take_one_netinfo_addr(r: &mut Reader<'_>) -> Result<Option<IpAddr>> {
         (_, _) => Ok(None),
     }
 }
+impl Netinfo {
+    /// Construct a new Netinfo to be sent by a client.
+    pub fn for_client(their_addr: IpAddr) -> Self {
+        Netinfo {
+            timestamp: 0, // clients don't report their timestamps.
+            their_addr,
+            my_addr: Vec::new(), // clients don't report their addrs.
+        }
+    }
+}
 impl Body for Netinfo {
     fn as_message(self) -> ChanMsg {
         ChanMsg::Netinfo(self)
@@ -470,6 +476,43 @@ impl Readable for Netinfo {
 pub struct Versions {
     versions: Vec<u16>,
 }
+impl Versions {
+    /// Construct a new Versions message using a provided list of link
+    /// protocols
+    pub fn new(vs: &[u16]) -> Self {
+        let versions = vs.into();
+        assert!(vs.len() < (std::u16::MAX / 2) as usize);
+        Self { versions }
+    }
+    /// Encode this VERSIONS cell in the manner expected for a handshake.
+    ///
+    /// (That's different from a standard cell encoding, since we
+    /// have not negotiated versions yet, and so our circuit-ID length
+    /// is an obsolete 2 bytes).
+    pub fn encode_for_handshake(self) -> Vec<u8> {
+        let mut v = Vec::new();
+        v.write_u16(0); // obsolete circuit ID length.
+        v.write_u8(ChanCmd::VERSIONS.into());
+        v.write_u16((self.versions.len() * 2) as u16); // message length.
+        self.write_body_onto(&mut v);
+        v
+    }
+    /// Return the best (numerically highest) link protocol that is
+    /// shared by this versions cell and my_protos.
+    pub fn best_shared_link_protocol(&self, my_protos: &[u16]) -> Option<u16> {
+        // NOTE: this implementation is quadratic, but it shouldn't matter
+        // much so long as my_protos is small.
+        let p = my_protos
+            .iter()
+            .filter(|p| self.versions.contains(p))
+            .fold(0u16, |a, b| u16::max(a, *b));
+        if p == 0 {
+            None
+        } else {
+            Some(p)
+        }
+    }
+}
 impl Body for Versions {
     fn as_message(self) -> ChanMsg {
         ChanMsg::Versions(self)

tor-proto/src/channel.rs

@@ -0,0 +1,156 @@
+//! Talking directly (over a TLS connection) to a Tor node
+
+#![allow(missing_docs)]
+
+use arrayref::array_ref;
+use futures::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use futures::sink::SinkExt;
+use futures::stream::StreamExt;
+
+use crate::chancell::{codec, msg, ChanCell, ChanCmd};
+use crate::{Error, Result};
+
+use std::net;
+use tor_bytes::Reader;
+use tor_linkspec::ChanTarget;
+
+// We only support version 4 for now, since we don't do padding right
+static LINK_PROTOCOLS: &[u16] = &[4];
+
+type CellFrame<T> = futures_codec::Framed<T, codec::ChannelCodec>;
+
+pub struct OutboundClientHandshake<T: AsyncRead + AsyncWrite + Unpin> {
+    tls: T,
+}
+
+pub struct UnverifiedChannel<T: AsyncRead + AsyncWrite + Unpin> {
+    link_protocol: u16,
+    tls: CellFrame<T>,
+    certs_cell: msg::Certs,
+    netinfo_cell: msg::Netinfo,
+}
+
+pub struct VerifiedChannel<T: AsyncRead + AsyncWrite + Unpin> {
+    link_protocol: u16,
+    tls: CellFrame<T>,
+}
+
+pub struct Channel<T: AsyncRead + AsyncWrite + Unpin> {
+    link_protocol: u16,
+    tls: CellFrame<T>,
+}
+
+impl<T: AsyncRead + AsyncWrite + Unpin> OutboundClientHandshake<T> {
+    pub fn new(tls: T) -> Self {
+        Self { tls }
+    }
+
+    pub async fn connect(mut self) -> Result<UnverifiedChannel<T>> {
+        // Send versions cell
+        {
+            let my_versions = msg::Versions::new(LINK_PROTOCOLS);
+            self.tls.write(&my_versions.encode_for_handshake()).await?;
+            self.tls.flush().await?;
+        }
+
+        // Get versions cell.
+        let their_versions: msg::Versions = {
+            let mut hdr = [0u8; 5];
+            self.tls.read(&mut hdr).await?;
+            if hdr[0..3] != [0, 0, ChanCmd::VERSIONS.into()] {
+                return Err(Error::ChanProto("Doesn't seem to be a tor relay".into()));
+            }
+            let msglen = u16::from_be_bytes(*array_ref![hdr, 3, 2]);
+            let mut msg = vec![0; msglen as usize];
+            self.tls.read_exact(&mut msg).await?;
+            let mut reader = Reader::from_slice(&msg);
+            reader.extract()?
+        };
+
+        // Determine shared versions.
+        let link_protocol = their_versions
+            .best_shared_link_protocol(LINK_PROTOCOLS)
+            .ok_or_else(|| Error::ChanProto("No shared link protocols".into()))?;
+
+        // Now we can switch to using a "Framed".
+        let mut tls = futures_codec::Framed::new(self.tls, codec::ChannelCodec::new(link_protocol));
+
+        let mut certs: Option<msg::Certs> = None;
+        let mut netinfo: Option<msg::Netinfo> = None;
+        let mut seen_authchallenge = false;
+
+        while let Some(m) = tls.next().await {
+            use msg::ChanMsg::*;
+            let (_, m) = m?.into_circid_and_msg();
+            // trace!("READ: {:?}", m);
+            match m {
+                // Are these technically allowed?
+                Padding(_) | VPadding(_) => (),
+                // Unrecognized cells get ignored.
+                Unrecognized(_) => (),
+                // Clients don't care about AuthChallenge
+                AuthChallenge(_) => {
+                    if seen_authchallenge {
+                        return Err(Error::ChanProto("Duplicate Authchallenge cell".into()));
+                    }
+                    seen_authchallenge = true;
+                }
+                Certs(c) => {
+                    if certs.is_some() {
+                        return Err(Error::ChanProto("Duplicate certs cell".into()));
+                    }
+                    certs = Some(c);
+                }
+                Netinfo(n) => {
+                    if netinfo.is_some() {
+                        return Err(Error::ChanProto("Duplicate certs cell".into()));
+                    }
+                    netinfo = Some(n);
+                    break;
+                }
+                // No other cell types are allowed.
+                m => {
+                    return Err(Error::ChanProto(format!(
+                        "Unexpected cell type {}",
+                        m.get_cmd()
+                    )))
+                }
+            }
+        }
+
+        match (certs, netinfo) {
+            (Some(_), None) => Err(Error::ChanProto("Missing netinfo or closed stream".into())),
+            (None, _) => Err(Error::ChanProto("Missing certs cell".into())),
+            (Some(certs_cell), Some(netinfo_cell)) => Ok(UnverifiedChannel {
+                link_protocol,
+                tls,
+                certs_cell,
+                netinfo_cell,
+            }),
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + Unpin> UnverifiedChannel<T> {
+    pub fn check<U: ChanTarget>(self, _peer: &U) -> Result<VerifiedChannel<T>> {
+        // XXXX need to verify certificates
+        Ok(VerifiedChannel {
+            link_protocol: self.link_protocol,
+            tls: self.tls,
+        })
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + Unpin> VerifiedChannel<T> {
+    pub async fn finish(mut self, peer_addr: &net::IpAddr) -> Result<Channel<T>> {
+        use msg::Body;
+        let netinfo = msg::Netinfo::for_client(*peer_addr);
+        let cell = ChanCell::new(0.into(), netinfo.as_message());
+        self.tls.send(cell).await?;
+
+        Ok(Channel {
+            link_protocol: self.link_protocol,
+            tls: self.tls,
+        })
+    }
+}

tor-proto/src/lib.rs

@@ -24,6 +24,7 @@
 #![deny(missing_docs)]
 
 pub mod chancell;
+pub mod channel;
 mod crypto;
 pub mod relaycell;
 mod util;

tor-proto/src/util/err.rs

@@ -36,7 +36,7 @@ pub enum Error {
     #[error("handshake failed")]
     BadHandshake,
     /// Protocol violation at the channel level
-    #[error("channel protocol violation")]
+    #[error("channel protocol violation: {0}")]
     ChanProto(String),
 }