//! Multi-hop paths over the Tor network. //! //! Right now, we only implement "client circuits" -- also sometimes //! called "origin circuits". A client circuit is one that is //! constructed by this Tor instance, and used in its own behalf to //! send data over the Tor network. //! //! Each circuit has multiple hops over the Tor network: each hop //! knows only the hop before and the hop after. The client shares a //! separate set of keys with each hop. //! //! To build a circuit, first create a [crate::channel::Channel], then //! call its [crate::channel::Channel::new_circ] method. This yields //! a [PendingClientCirc] object that won't become live until you call //! one of the methods that extends it to its first hop. After you've //! done that, you can call [ClientCirc::extend_ntor] on the circuit to //! build it into a multi-hop circuit. Finally, you can use //! [ClientCirc::begin_stream] to get a Stream object that can be used //! for anonymized data. //! //! # Implementation //! //! Each open circuit has a corresponding Reactor object that runs in //! an asynchronous task, and manages incoming cells from the //! circuit's upstream channel. These cells are either RELAY cells or //! DESTROY cells. DESTROY cells are handled immediately. //! RELAY cells are either for a particular stream, in which case they //! get forwarded to a RawCellStream object, or for no particular stream, //! in which case they are considered "meta" cells (like EXTENEDED2) //! that should only get accepted if something is waiting for them. //! //! # Limitations //! //! This is client-only. //! //! There's one big mutex on the whole circuit: the reactor needs to hold //! it to process a cell, and streams need to hold it to send. //! //! There is no flow-control or rate-limiting or fairness. pub(crate) mod celltypes; pub(crate) mod halfcirc; mod halfstream; pub(crate) mod reactor; pub(crate) mod sendme; mod streammap; mod unique_id; use crate::channel::{Channel, CircDestroyHandle}; use crate::circuit::celltypes::*; use crate::circuit::reactor::{CtrlMsg, CtrlResult}; pub use crate::circuit::unique_id::UniqId; use crate::crypto::cell::{ ClientLayer, CryptInit, HopNum, InboundClientLayer, OutboundClientCrypt, OutboundClientLayer, RelayCellBody, }; use crate::crypto::handshake::{ClientHandshake, KeyGenerator}; use crate::stream::{DataStream, RawCellStream}; use crate::{Error, Result}; use tor_cell::chancell::{self, msg::ChanMsg, ChanCell, CircId}; use tor_cell::relaycell::msg::{RelayMsg, Sendme}; use tor_cell::relaycell::{RelayCell, RelayCmd, StreamId}; use tor_linkspec::{ChanTarget, CircTarget, LinkSpec}; pub use tor_cell::relaycell::msg::IpVersionPreference; use futures::channel::{mpsc, oneshot}; use futures::lock::Mutex; use futures::sink::SinkExt; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; // use std::time::Duration; use rand::{thread_rng, CryptoRng, Rng}; use log::{debug, trace}; /// A circuit that we have constructed over the Tor network. pub struct ClientCirc { /// This circuit can't be used because it has been closed, locally /// or remotely. closed: AtomicBool, /// A unique identifier for this circuit. unique_id: UniqId, /// Reference-counted locked reference to the inner circuit object. c: Mutex, } impl std::fmt::Debug for ClientCirc { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ClientCirc") .field("unique_id", &self.unique_id) .field("closed", &self.closed) .finish() } } /// A ClientCirc that needs to send a create cell and receive a created* cell. /// /// To use one of these, call create_firsthop_fast() or create_firsthop_ntor() /// to negotiate the cryptographic handshake with the first hop. pub struct PendingClientCirc { /// A oneshot receiver on which we'll receive a CREATED* cell, /// or a DESTROY cell. recvcreated: oneshot::Receiver, /// The ClientCirc object that we can expose on success. circ: Arc, } /// Description of the network's current rules for building circuits. #[derive(Clone, Debug)] pub struct CircParameters { /// Initial value to use for our outbound circuit-level windows. initial_send_window: u16, /// Whether we should include ed25519 identities when we send /// EXTEND2 cells. extend_by_ed25519_id: bool, } impl Default for CircParameters { fn default() -> CircParameters { CircParameters { initial_send_window: 1000, extend_by_ed25519_id: true, } } } impl CircParameters { /// Override the default initial send window for these parameters. /// Gives an error on any value above 1000. /// /// You should probably not call this. pub fn set_initial_send_window(&mut self, v: u16) -> Result<()> { if v <= 1000 { self.initial_send_window = v; Ok(()) } else { Err(Error::BadConfig( "Tried to set an initial send window over 1000".into(), )) } } /// Return the initial send window as set in this parameter set. pub fn initial_send_window(&self) -> u16 { self.initial_send_window } /// Override the default decision about whether to use ed25519 /// identities in outgoing EXTEND2 cells. /// /// You should probably not call this. pub fn set_extend_by_ed25519_id(&mut self, v: bool) { self.extend_by_ed25519_id = v; } /// Return true if we're configured to extend by ed25519 ID; false /// otherwise. pub fn extend_by_ed25519_id(&self) -> bool { self.extend_by_ed25519_id } } /// A result type used to tell a circuit about some a "meta-cell" /// (like extended, intro_established, etc). type MetaResult = Result; /// The implementation type for this circuit. struct ClientCircImpl { /// This circuit's ID on the upstream channel. id: CircId, /// The channel that this circuit uses to send its cells to the /// next hop. channel: Arc, /// The cryptographic state for this circuit for outbound cells. /// This object is divided into multiple layers, each of which is /// shared with one hop of the circuit crypto_out: OutboundClientCrypt, /// When this is dropped, the channel reactor is told to send a DESTROY /// cell. circ_closed: Option, /// Per-hop circuit information. /// /// Note that hops.len() must be the same as crypto.n_layers(). hops: Vec, /// A stream that can be used to register streams with the reactor. control: mpsc::Sender, /// A oneshot sender that can be used to tell the reactor to shut down. sendshutdown: Option>, /// A oneshot sender that can be used by the reactor to report a /// meta-cell to an owning task. /// /// This comes along with a hop number saying which hop we expect a /// meta-cell from. Cells from other hops won't go to this sender. /// /// For the purposes of this implementation, a "meta" cell /// is a RELAY cell with a stream ID value of 0. sendmeta: Option<(HopNum, oneshot::Sender)>, /// An identifier for this circuit, for logging purposes. /// TODO: Make this field go away in favor of the one in ClientCirc. unique_id: UniqId, } /// A handle to a circuit as held by a stream. Used to send cells. /// /// Rather than using the stream directly, the stream uses this object /// to send its relay cells to the correct hop, using the correct stream ID. /// /// When this object is dropped, the reactor will be told to close the stream. // XXXX TODO: rename this pub(crate) struct StreamTarget { /// The stream ID for this stream on its circuit. stream_id: StreamId, /// Which hop on this circuit is this stream built from? // XXXX Using 'hop' by number here will cause bugs if circuits can get // XXXX truncated and then re-extended. hop: HopNum, /// Reference to the circuit that this stream is on. circ: Arc, /// Window for sending cells on this circuit. window: sendme::StreamSendWindow, /// One-shot sender that should get a message once this stream /// is dropped. stream_closed: Option>, /// Window to track incoming cells and SENDMEs. // XXXX Putting this field here in this object means that this // object isn't really so much a "target", since a "target" // doesn't know how to receive. Maybe we should rename it to be // some kind of a "handle" or something? pub(crate) recvwindow: sendme::StreamRecvWindow, } /// Information about a single hop of a client circuit, from the sender-side /// point of view. /// /// (see also circuit::reactor::InboundHop) struct CircHop { /// If true, this hop is using an older link protocol and we /// shouldn't expect good authenticated SENDMEs from it. auth_sendme_optional: bool, /// Window used to say how many cells we can send. sendwindow: sendme::CircSendWindow, } impl CircHop { /// Construct a new (sender-side) view of a circuit hop. fn new(auth_sendme_optional: bool, initial_window: u16) -> Self { CircHop { auth_sendme_optional, sendwindow: sendme::CircSendWindow::new(initial_window), } } } impl ClientCirc { /// Helper: return the number of hops for this circuit #[cfg(test)] async fn n_hops(&self) -> usize { let c = self.c.lock().await; c.crypto_out.n_layers() } /// Helper: extend the circuit by one hop. /// /// The `rng` is used to generate handshake material. The /// `handshake_id` is the numeric identifer for what kind of /// handshake we're doing. The `key is the relay's onion key that /// goes along with the handshake, and the `linkspecs` are the /// link specifiers to include in the EXTEND cell to tell the /// current last hop which relay to connect to. async fn extend_impl( &self, rng: &mut R, handshake_id: u16, key: &H::KeyType, linkspecs: Vec, supports_flowctrl_1: bool, params: &CircParameters, ) -> Result<()> where R: Rng + CryptoRng, L: CryptInit + ClientLayer, FWD: OutboundClientLayer + 'static + Send, REV: InboundClientLayer + 'static + Send, H: ClientHandshake, H::KeyGen: KeyGenerator, { use tor_cell::relaycell::msg::{Body, Extend2}; // Perform the first part of the cryptographic handshake let (state, msg) = H::client1(rng, key)?; // Cloning linkspecs is only necessary because of the log // below. Would be nice to fix that. let extend_msg = Extend2::new(linkspecs.clone(), handshake_id, msg); let cell = RelayCell::new(0.into(), extend_msg.into_message()); // Now send the EXTEND2 cell to the the last hop... let (unique_id, _hop, receiver) = { let mut c = self.c.lock().await; let n_hops = c.crypto_out.n_layers(); let hop = ((n_hops - 1) as u8).into(); debug!( "{}: Extending circuit to hop {} with {:?}", c.unique_id, n_hops + 1, linkspecs ); // We'll be waiting for an EXTENDED2 cell; install the handler. let receiver = c.register_meta_handler(hop)?; // Send the message to the last hop... c.send_relay_cell( hop, true, // use a RELAY_EARLY cell cell, ) .await?; (c.unique_id, hop, receiver) // note that we're dropping the lock here, since we're going // to wait for a response. }; trace!("{}: waiting for EXTENDED2 cell", unique_id); // ... and now we wait for a response. let msg = match receiver.await { Ok(Ok(m)) => Ok(m), Err(_) => Err(Error::InternalError( "Receiver cancelled while waiting for EXTENDED2".into(), )), Ok(Err(Error::CircuitClosed)) => Err(Error::CircDestroy( "Circuit closed while waiting for EXTENDED2".into(), )), Ok(Err(e)) => Err(e), }?; // XXXX If two EXTEND cells are of these are launched on the // same circuit at once, could they collide in this part of // the function? I don't _think_ so, but it might be a good idea // to have an "extending" bit that keeps two tasks from entering // extend_impl at the same time. // // Also we could enforce that `hop` is still what we expect it // to be at this point. // Did we get the right response? if msg.cmd() != RelayCmd::EXTENDED2 { self.protocol_error().await; return Err(Error::CircProto(format!( "wanted EXTENDED2; got {}", msg.cmd(), ))); } // ???? Do we need to shutdown the circuit for the remaining error // ???? cases in this function? let msg = match msg { RelayMsg::Extended2(e) => e, _ => return Err(Error::InternalError("Body didn't match cmd".into())), }; let relay_handshake = msg.into_body(); trace!( "{}: Received EXTENDED2 cell; completing handshake.", unique_id ); // Now perform the second part of the handshake, and see if it // succeeded. let keygen = H::client2(state, relay_handshake)?; let layer = L::construct(keygen)?; debug!("{}: Handshake complete; circuit extended.", unique_id); // If we get here, it succeeded. Add a new hop to the circuit. let (layer_fwd, layer_back) = layer.split(); self.add_hop( supports_flowctrl_1, Box::new(layer_fwd), Box::new(layer_back), params, ) .await } /// Add a hop to the end of this circuit. /// /// This function is a bit tricky, since we need to add the /// hop to our own structures, and tell the reactor to add it to the /// reactor's structures as well, and wait for the reactor to tell us /// that it did. async fn add_hop<'a>( &'a self, supports_flowctrl_1: bool, fwd: Box, rev: Box, params: &'a CircParameters, ) -> Result<()> { let inbound_hop = crate::circuit::reactor::InboundHop::new(); let (snd, rcv) = oneshot::channel(); { let mut c = self.c.lock().await; c.control .send(Ok(CtrlMsg::AddHop(inbound_hop, rev, snd))) .await .map_err(|_| Error::InternalError("Can't queue AddHop request".into()))?; } // I think we don't need to worry about two hops being added at // once, because there can only be on meta-message receiver at // a time. rcv.await .map_err(|_| Error::InternalError("AddHop request cancelled".into()))?; { let mut c = self.c.lock().await; let hop = CircHop::new(supports_flowctrl_1, params.initial_send_window()); c.hops.push(hop); c.crypto_out.add_layer(fwd); } Ok(()) } /// Extend the circuit via the ntor handshake to a new target last /// hop. /// /// The same caveats apply from extend_impl. pub async fn extend_ntor( &self, rng: &mut R, target: &Tg, params: &CircParameters, ) -> Result<()> where R: Rng + CryptoRng, Tg: CircTarget, { use crate::crypto::cell::Tor1RelayCrypto; use crate::crypto::handshake::ntor::{NtorClient, NtorPublicKey}; let key = NtorPublicKey { id: *target.rsa_identity(), pk: *target.ntor_onion_key(), }; let mut linkspecs = target.linkspecs(); if !params.extend_by_ed25519_id() { linkspecs.retain(|ls| !matches!(ls, LinkSpec::Ed25519Id(_))); } // FlowCtrl=1 means that this hop supports authenticated SENDMEs let supports_flowctrl_1 = target .protovers() .supports_known_subver(tor_protover::ProtoKind::FlowCtrl, 1); self.extend_impl::( rng, 0x0002, &key, linkspecs, supports_flowctrl_1, params, ) .await } /// Helper, used to begin a stream. /// /// This function allocates a stream ID, and sends the message /// (like a BEGIN or RESOLVE), but doesn't wait for a response. /// /// The caller will typically want to see the first cell in response, /// to see whether it is e.g. an END or a CONNECTED. async fn begin_stream_impl(self: &Arc, begin_msg: RelayMsg) -> Result { // TODO: Possibly this should take a hop, rather than just // assuming it's the last hop. // XXXX Both a bound and a lack of bound are scary here :/ let (sender, receiver) = mpsc::channel(128); let (send_close, recv_close) = oneshot::channel::(); let window = sendme::StreamSendWindow::new(StreamTarget::SEND_WINDOW_INIT); let (id_snd, id_rcv) = oneshot::channel(); let hopnum; { let mut c = self.c.lock().await; let h = c.hops.len() - 1; hopnum = (h as u8).into(); c.control .send(Ok(CtrlMsg::AddStream( hopnum, sender, window.new_ref(), id_snd, ))) .await .map_err(|_| Error::InternalError("Can't queue new-stream request.".into()))?; } let id = id_rcv .await .map_err(|_| Error::InternalError("Didn't receive a stream ID.".into()))?; let id = id?; let relaycell = RelayCell::new(id, begin_msg); { let mut c = self.c.lock().await; c.send_relay_cell(hopnum, false, relaycell).await?; c.control .send(Ok(CtrlMsg::Register(recv_close))) .await .map_err(|_| Error::InternalError("Can't queue stream closer".into()))?; } /// Initial value for inbound flow-control window on streams. const STREAM_RECV_INIT: u16 = 500; let target = StreamTarget { circ: Arc::clone(self), stream_id: id, hop: hopnum, window, recvwindow: sendme::StreamRecvWindow::new(STREAM_RECV_INIT), stream_closed: Some(send_close), }; Ok(RawCellStream::new(target, receiver)) } /// Start a DataStream (anonymized connection) to the given /// address and port, using a BEGIN cell. async fn begin_data_stream(self: Arc, msg: RelayMsg) -> Result { let stream = self.begin_stream_impl(msg).await?; // TODO: waiting for a response here preculdes optimistic data. let response = stream.recv().await?; match response { RelayMsg::Connected(_) => Ok(DataStream::new(stream)), RelayMsg::End(cell) => Err(Error::EndReceived(cell.reason())), _ => { self.protocol_error().await; Err(Error::StreamProto(format!( "Received {} while waiting for connection", response.cmd() ))) } } } /// Start a stream to the given address and port, using a BEGIN /// cell. /// /// The use of a string for the address is intentional: you should let /// the remote Tor relay do the hostname lookup for you. pub async fn begin_stream( self: Arc, target: &str, port: u16, flags: Option, ) -> Result { let flags = flags.unwrap_or_default(); let beginmsg = tor_cell::relaycell::msg::Begin::new(target, port, flags)?; self.begin_data_stream(beginmsg.into()).await } /// Start a new stream to the last relay in the circuit, using /// a BEGIN_DIR cell. pub async fn begin_dir_stream(self: Arc) -> Result { self.begin_data_stream(RelayMsg::BeginDir).await } // XXXX Add a RESOLVE implementation, it will be simple. /// Helper: Encode the relay cell `cell`, encrypt it, and send it to the /// 'hop'th hop. /// /// Does not check whether the cell is well-formed or reasonable. async fn send_relay_cell(&self, hop: HopNum, early: bool, cell: RelayCell) -> Result<()> { if self.closed.load(Ordering::SeqCst) { return Err(Error::CircuitClosed); } let mut c = self.c.lock().await; c.send_relay_cell(hop, early, cell).await } /// Shut down this circuit immediately, along with all streams that /// are using it. /// /// Note that other references to this circuit may exist. If they /// do, they will stop working after you call this function. /// /// It's not necessary to call this method if you're just done /// with a circuit: the channel should close on its own once nothing /// is using it any more. pub async fn terminate(&self) { let outcome = self .closed .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst); if outcome == Ok(false) { // The old value was false and the new value is true. self.c.lock().await.shutdown_reactor(); } } /// Called when a circuit-level protocol error has occurred and the /// circuit needs to shut down. /// /// This is a separate function because we may eventually want to have /// it do more than just shut down. pub(crate) async fn protocol_error(&self) { self.terminate().await; } /// Return true if this circuit is closed and therefore unusable. pub fn is_closing(&self) -> bool { self.closed.load(Ordering::SeqCst) } /// Return a process-unique identifier for this circui. pub fn unique_id(&self) -> UniqId { self.unique_id } /// Helper: register a meta-handler for this circuit. #[cfg(test)] async fn register_meta_handler(&self, hop: HopNum) -> Result> { let mut c = self.c.lock().await; c.register_meta_handler(hop) } } impl ClientCircImpl { /// Return a mutable reference to the nth hop of this circuit, if one /// exists. fn hop_mut(&mut self, hopnum: HopNum) -> Option<&mut CircHop> { self.hops.get_mut(Into::::into(hopnum)) } /// Helper: Register a handler that will be told about the RELAY message /// with StreamId 0. /// /// This pattern is useful for parts of the protocol where the circuit /// originator sends a single request, and waits for a single relay /// message in response. (For example, EXTEND/EXTENDED, /// ESTABLISH_RENDEZVOUS/RENDEZVOUS_ESTABLISHED, and so on.) /// /// It isn't suitable for SENDME cells, INTRODUCE2 cells, or TRUNCATED /// cells. /// /// Only one handler can be registered at a time; until it fires or is /// cancelled, you can't register another. /// /// Note that you should register a meta handler _before_ you send whatever /// cell you're waiting a response to, or you might miss the response. // TODO: It would be cool for this to take a list of allowable // cell types to get in response, so that any other cell types are // treated as circuit protocol violations automatically. fn register_meta_handler(&mut self, hop: HopNum) -> Result> { // Was there previously a handler? if self.sendmeta.is_some() { return Err(Error::InternalError( "Tried to register a second meta-cell handler".into(), )); } let (sender, receiver) = oneshot::channel(); self.sendmeta = Some((hop, sender)); trace!( "{}: Registered a meta-cell handler for hop {}", self.unique_id, hop ); Ok(receiver) } /// Handle a RELAY cell on this circuit with stream ID 0. async fn handle_meta_cell(&mut self, hopnum: HopNum, msg: RelayMsg) -> Result<()> { // SENDME cells and TRUNCATED get handled internally by the circuit. if let RelayMsg::Sendme(s) = msg { return self.handle_sendme(hopnum, s).await; } if let RelayMsg::Truncated(_) = msg { // XXXX need to handle Truncated cells. This isn't the right // way, but at least it's safe. // TODO: If we ever do handle Truncate cells more // correctly, we will need to audit all our use of HopNum // to identify a layer. Otherwise we could confuse a // message from the previous hop N with a message from the // new hop N. return Err(Error::CircuitClosed); } trace!("{}: Received meta-cell {:?}", self.unique_id, msg); // For all other command types, we'll only get them in response // to another command, which should have registered a responder. // // TODO: that means that service-introduction circuits will need // a different implementation, but that should be okay. We'll work // something out. if let Some((expected_hop, sender)) = self.sendmeta.take() { if expected_hop == hopnum { // Somebody was waiting for a message -- maybe this message sender .send(Ok(msg)) // I think this means that the channel got closed. .map_err(|_| Error::CircuitClosed) } else { // Somebody wanted a message from a different hop! Put this // one back. self.sendmeta = Some((expected_hop, sender)); Err(Error::CircProto(format!( "Unexpected {} cell from hop {} on client circuit", msg.cmd(), hopnum, ))) } } else { // No need to call shutdown here, since this error will // propagate to the reactor shut it down. Err(Error::CircProto(format!( "Unexpected {} cell on client circuit", msg.cmd() ))) } } /// Handle a RELAY_SENDME cell on this circuit with stream ID 0. async fn handle_sendme(&mut self, hopnum: HopNum, msg: Sendme) -> Result<()> { // No need to call "shutdown" on errors in this function; // it's called from the reactor task and errors will propagate there. let hop = self .hop_mut(hopnum) .ok_or_else(|| Error::CircProto(format!("Couldn't find {} hop", hopnum)))?; let auth: Option<[u8; 20]> = match msg.into_tag() { Some(v) if v.len() == 20 => { // XXXX ugly code. let mut tag = [0_u8; 20]; (&mut tag).copy_from_slice(&v[..]); Some(tag) } Some(_) => return Err(Error::CircProto("malformed tag on circuit sendme".into())), None => { if !hop.auth_sendme_optional { return Err(Error::CircProto("missing tag on circuit sendme".into())); } else { None } } }; match hop.sendwindow.put(auth).await { Some(_) => Ok(()), None => Err(Error::CircProto("bad auth tag on circuit sendme".into())), } } /// Helper: Put a cell onto this circuit's channel. /// /// This takes a raw cell that has already been encrypted, puts /// a circuit ID on it, and sends it. /// /// Does not check whether the cell is well-formed or reasonable. async fn send_msg(&mut self, msg: ChanMsg) -> Result<()> { let cell = ChanCell::new(self.id, msg); self.channel.send_cell(cell).await?; Ok(()) } /// Helper: Encode the relay cell `cell`, encrypt it, and send it to the /// 'hop'th hop. /// /// Does not check whether the cell is well-formed or reasonable. async fn send_relay_cell(&mut self, hop: HopNum, early: bool, cell: RelayCell) -> Result<()> { let c_t_w = sendme::cell_counts_towards_windows(&cell); let mut body: RelayCellBody = cell.encode(&mut thread_rng())?.into(); let tag = self.crypto_out.encrypt(&mut body, hop)?; let msg = chancell::msg::Relay::from_raw(body.into()); let msg = if early { ChanMsg::RelayEarly(msg) } else { ChanMsg::Relay(msg) }; // If the cell counted towards our sendme window, decrement // that window, and maybe remember the authentication tag. if c_t_w { // TODO: I'd like to use get_hops_mut here, but the borrow checker // won't let me. // This blocks if the send window is empty. self.hops[Into::::into(hop)] .sendwindow .take(tag) .await?; } self.send_msg(msg).await } /// Shut down this circuit's reactor and send a DESTROY cell. /// /// This is idempotent and safe to call more than once. fn shutdown_reactor(&mut self) { if let Some(sender) = self.sendshutdown.take() { // ignore the error, since it can only be canceled. let _ = sender.send(CtrlMsg::Shutdown); } // Drop the circuit destroy handle now so that a DESTROY cell // gets sent. drop(self.circ_closed.take()); } } impl PendingClientCirc { /// Instantiate a new circuit object: used from Channel::new_circ(). /// /// Does not send a CREATE* cell on its own. /// /// pub(crate) fn new( id: CircId, channel: Arc, createdreceiver: oneshot::Receiver, circ_closed: Option, input: mpsc::Receiver, unique_id: UniqId, ) -> (PendingClientCirc, reactor::Reactor) { let crypto_out = OutboundClientCrypt::new(); let (sendclosed, recvclosed) = oneshot::channel::(); // Should this be bounded, really? XXX let (sendctrl, recvctrl) = mpsc::channel::(128); let hops = Vec::new(); let circuit_impl = ClientCircImpl { id, channel, crypto_out, hops, circ_closed, control: sendctrl, sendshutdown: Some(sendclosed), sendmeta: None, unique_id, }; let circuit = ClientCirc { closed: AtomicBool::new(false), c: Mutex::new(circuit_impl), unique_id, }; let circuit = Arc::new(circuit); let pending = PendingClientCirc { recvcreated: createdreceiver, circ: Arc::clone(&circuit), }; let reactor = reactor::Reactor::new(circuit, recvctrl, recvclosed, input, unique_id); (pending, reactor) } /// Check whether this pending circuit matches a given channel target; /// return an error if it doesn't. async fn check_chan_match(&self, target: &T) -> Result<()> { let c = self.circ.c.lock().await; c.channel.check_match(target) } /// Testing only: extract the circuit ID for thid pending circuit. #[cfg(test)] pub(crate) async fn peek_circid(&self) -> CircId { let c = self.circ.c.lock().await; c.id } /// Helper: create the first hop of a circuit. /// /// This is parameterized not just on the RNG, but a wrapper object to /// build the right kind of create cell, a handshake object to perform /// the cryptographic cryptographic handshake, and a layer type to /// handle relay crypto after this hop is built. async fn create_impl( self, rng: &mut R, wrap: &W, key: &H::KeyType, supports_flowctrl_1: bool, params: &CircParameters, ) -> Result> where R: Rng + CryptoRng, L: CryptInit + ClientLayer + 'static + Send, // need all this?XXXX FWD: OutboundClientLayer + 'static + Send, REV: InboundClientLayer + 'static + Send, H: ClientHandshake, W: CreateHandshakeWrap, H::KeyGen: KeyGenerator, { // We don't need to shut down the circuit on failure here, since this // function consumes the PendingClientCirc and only returns // a ClientCirc on success. let PendingClientCirc { circ, recvcreated } = self; let (state, msg) = H::client1(rng, key)?; let create_cell = wrap.to_chanmsg(msg); let unique_id = { let mut c = circ.c.lock().await; debug!( "{}: Extending to hop 1 with {}", c.unique_id, create_cell.cmd() ); c.send_msg(create_cell).await?; c.unique_id }; let reply = recvcreated .await .map_err(|_| Error::CircProto("Circuit closed while waiting".into()))?; let relay_handshake = wrap.from_chanmsg(reply)?; let keygen = H::client2(state, relay_handshake)?; let layer = L::construct(keygen)?; debug!("{}: Handshake complete; circuit created.", unique_id); let (layer_fwd, layer_back) = layer.split(); circ.add_hop( supports_flowctrl_1, Box::new(layer_fwd), Box::new(layer_back), params, ) .await?; Ok(circ) } /// Use the (questionable!) CREATE_FAST handshake to connect to the /// first hop of this circuit. /// /// There's no authentication in CRATE_FAST, /// so we don't need to know whom we're connecting to: we're just /// connecting to whichever relay the channel is for. pub async fn create_firsthop_fast( self, rng: &mut R, params: &CircParameters, ) -> Result> where R: Rng + CryptoRng, { use crate::crypto::cell::Tor1RelayCrypto; use crate::crypto::handshake::fast::CreateFastClient; let wrap = CreateFastWrap; self.create_impl::( rng, &wrap, &(), false, params, ) .await } /// Use the ntor handshake to connect to the first hop of this circuit. /// /// Note that the provided 'target' must match the channel's target, /// or the handshake will fail. pub async fn create_firsthop_ntor( self, rng: &mut R, target: &Tg, params: &CircParameters, ) -> Result> where R: Rng + CryptoRng, Tg: tor_linkspec::CircTarget, { use crate::crypto::cell::Tor1RelayCrypto; use crate::crypto::handshake::ntor::{NtorClient, NtorPublicKey}; // Exit now if we have an Ed25519 or RSA identity mismatch. self.check_chan_match(target).await?; let wrap = Create2Wrap { handshake_type: 0x0002, // ntor }; let key = NtorPublicKey { id: *target.rsa_identity(), pk: *target.ntor_onion_key(), }; // FlowCtrl=1 means that this hop supports authenticated SENDMEs let supports_flowctrl_1 = target .protovers() .supports_known_subver(tor_protover::ProtoKind::FlowCtrl, 1); self.create_impl::( rng, &wrap, &key, supports_flowctrl_1, params, ) .await } } /// An object that can put a given handshake into a ChanMsg for a CREATE* /// cell, and unwrap a CREATED* cell. trait CreateHandshakeWrap { /// Construct an appropriate ChanMsg to hold this kind of handshake. fn to_chanmsg(&self, bytes: Vec) -> ChanMsg; /// Decode a ChanMsg to an appropriate handshake value, checking /// its type. fn from_chanmsg(&self, msg: CreateResponse) -> Result>; } /// A CreateHandshakeWrap that generates CREATE_FAST and handles CREATED_FAST. struct CreateFastWrap; impl CreateHandshakeWrap for CreateFastWrap { fn to_chanmsg(&self, bytes: Vec) -> ChanMsg { chancell::msg::CreateFast::new(bytes).into() } fn from_chanmsg(&self, msg: CreateResponse) -> Result> { use CreateResponse::*; match msg { CreatedFast(m) => Ok(m.into_body()), Destroy(_) => Err(Error::CircExtend( "Relay replied to CREATE_FAST with DESTROY.", )), _ => Err(Error::CircExtend( "Relay replied to CREATE_FAST with unexpected cell.", )), } } } /// A CreateHandshakeWrap that generates CREATE2 and handles CREATED2 struct Create2Wrap { /// The handshake type to put in the CREATE2 cell. handshake_type: u16, } impl CreateHandshakeWrap for Create2Wrap { fn to_chanmsg(&self, bytes: Vec) -> ChanMsg { chancell::msg::Create2::new(self.handshake_type, bytes).into() } fn from_chanmsg(&self, msg: CreateResponse) -> Result> { use CreateResponse::*; match msg { Created2(m) => Ok(m.into_body()), Destroy(_) => Err(Error::CircExtend("Relay replied to CREATE2 with DESTROY.")), _ => Err(Error::CircExtend( "Relay replied to CREATE2 with unexpected cell.", )), } } } impl StreamTarget { /// Initial value for outbound flow-control window on streams. const SEND_WINDOW_INIT: u16 = 500; /// Deliver a relay message for the stream that owns this StreamTarget. /// /// The StreamTarget will set the correct stream ID and pick the /// right hop, but will not validate that the message is well-formed /// or meaningful in context. pub(crate) async fn send(&mut self, msg: RelayMsg) -> Result<()> { if sendme::msg_counts_towards_windows(&msg) { // Decrement the stream window (and block if it's empty) self.window.take(&()).await?; } let cell = RelayCell::new(self.stream_id, msg); self.circ.send_relay_cell(self.hop, false, cell).await } /// Called when a circuit-level protocol error has occurred and the /// circuit needs to shut down. pub(crate) async fn protocol_error(&mut self) { self.circ.protocol_error().await; } } impl Drop for ClientCircImpl { fn drop(&mut self) { self.shutdown_reactor(); } } impl Drop for StreamTarget { fn drop(&mut self) { if let Some(sender) = self.stream_closed.take() { // This "clone" call is a bit dangerous: it means that we might // allow the other side to send a couple of cells that get // decremented from self.recvwindow but don't get reflected // in the circuit-owned view of the window. let window = self.recvwindow.clone(); let _ = sender.send(CtrlMsg::CloseStream(self.hop, self.stream_id, window)); } // If there's an error, no worries: it's hard-cancel, and we // can just ignore it. XXXX (I hope?) } } #[cfg(test)] mod test { use super::*; use crate::channel::test::fake_channel; use chanmsg::{ChanMsg, Created2, CreatedFast}; use futures::io::{AsyncReadExt, AsyncWriteExt}; use futures::stream::StreamExt; use futures_await_test::async_test; use hex_literal::hex; use tor_cell::chancell::msg as chanmsg; use tor_cell::relaycell::msg as relaymsg; use tor_llcrypto::pk; fn rmsg_to_ccmsg(id: ID, msg: relaymsg::RelayMsg) -> ClientCircChanMsg where ID: Into, { let body: RelayCellBody = RelayCell::new(id.into(), msg) .encode(&mut thread_rng()) .unwrap() .into(); let chanmsg = chanmsg::Relay::from_raw(body.into()); ClientCircChanMsg::Relay(chanmsg) } struct ExampleTarget { ntor_key: pk::curve25519::PublicKey, protovers: tor_protover::Protocols, ed_id: pk::ed25519::Ed25519Identity, rsa_id: pk::rsa::RsaIdentity, } impl tor_linkspec::ChanTarget for ExampleTarget { fn addrs(&self) -> &[std::net::SocketAddr] { &[] } fn ed_identity(&self) -> &pk::ed25519::Ed25519Identity { &self.ed_id } fn rsa_identity(&self) -> &pk::rsa::RsaIdentity { &self.rsa_id } } impl tor_linkspec::CircTarget for ExampleTarget { fn ntor_onion_key(&self) -> &pk::curve25519::PublicKey { &self.ntor_key } fn protovers(&self) -> &tor_protover::Protocols { &self.protovers } } /// return an ExampleTarget that can get used for an ntor handshake. fn example_target() -> ExampleTarget { ExampleTarget { ntor_key: hex!("395cb26b83b3cd4b91dba9913e562ae87d21ecdd56843da7ca939a6a69001253") .into(), protovers: "FlowCtrl=1".parse().unwrap(), ed_id: [6_u8; 32].into(), rsa_id: [10_u8; 20].into(), } } fn example_ntor_key() -> crate::crypto::handshake::ntor::NtorSecretKey { crate::crypto::handshake::ntor::NtorSecretKey::new( hex!("7789d92a89711a7e2874c61ea495452cfd48627b3ca2ea9546aafa5bf7b55803").into(), hex!("395cb26b83b3cd4b91dba9913e562ae87d21ecdd56843da7ca939a6a69001253").into(), [10_u8; 20].into(), ) } async fn test_create(fast: bool) { // We want to try progressing from a pending circuit to a circuit // via a crate_fast handshake. use crate::crypto::handshake::{fast::CreateFastServer, ntor::NtorServer, ServerHandshake}; use futures::future::FutureExt; let (chan, mut ch) = fake_channel(); let circid = 128.into(); let (created_send, created_recv) = oneshot::channel(); let (_circmsg_send, circmsg_recv) = mpsc::channel(64); let unique_id = UniqId::new(23, 17); let (pending, mut reactor) = PendingClientCirc::new( circid, chan, created_recv, None, // circ_closed. circmsg_recv, unique_id, ); // Future to pretend to be a relay on the other end of the circuit. let simulate_relay_fut = async move { let mut rng = rand::thread_rng(); let create_cell = ch.cells.next().await.unwrap(); assert_eq!(create_cell.circid(), 128.into()); let reply = if fast { let cf = match create_cell.msg() { ChanMsg::CreateFast(cf) => cf, _ => panic!(), }; let (_, rep) = CreateFastServer::server(&mut rng, &[()], cf.body()).unwrap(); CreateResponse::CreatedFast(CreatedFast::new(rep)) } else { let c2 = match create_cell.msg() { ChanMsg::Create2(c2) => c2, _ => panic!(), }; let (_, rep) = NtorServer::server(&mut rng, &[example_ntor_key()], c2.body()).unwrap(); CreateResponse::Created2(Created2::new(rep)) }; created_send.send(reply).unwrap(); }; // Future to pretend to be a client. let client_fut = async move { let mut rng = rand::thread_rng(); let target = example_target(); let params = CircParameters::default(); if fast { pending.create_firsthop_fast(&mut rng, ¶ms).await } else { pending .create_firsthop_ntor(&mut rng, &target, ¶ms) .await } }; // Future to run the reactor. let reactor_fut = reactor.run_once().map(|_| ()); let (circ, _, _) = futures::join!(client_fut, reactor_fut, simulate_relay_fut); let _circ = circ.unwrap(); // pfew! We've build a circuit! Let's make sure it has one hop. /* TODO: reinstate this. let inner = Arc::get_mut(&mut circuit).unwrap().c.into_inner(); assert_eq!(inner.hops.len(), 1); */ } #[async_test] async fn test_create_fast() { test_create(true).await } #[async_test] async fn test_create_ntor() { test_create(false).await } // An encryption layer that doesn't do any crypto. Can be used // as inbound or outbound, but not both at once. struct DummyCrypto { counter_tag: [u8; 20], counter: u32, lasthop: bool, } impl DummyCrypto { fn next_tag(&mut self) -> &[u8; 20] { self.counter_tag[0] = ((self.counter >> 0) & 255) as u8; self.counter_tag[1] = ((self.counter >> 8) & 255) as u8; self.counter_tag[2] = ((self.counter >> 16) & 255) as u8; self.counter_tag[3] = ((self.counter >> 24) & 255) as u8; self.counter += 1; &self.counter_tag } } impl crate::crypto::cell::OutboundClientLayer for DummyCrypto { fn originate_for(&mut self, _cell: &mut RelayCellBody) -> &[u8] { self.next_tag() } fn encrypt_outbound(&mut self, _cell: &mut RelayCellBody) {} } impl crate::crypto::cell::InboundClientLayer for DummyCrypto { fn decrypt_inbound(&mut self, _cell: &mut RelayCellBody) -> Option<&[u8]> { if self.lasthop { Some(self.next_tag()) } else { None } } } impl DummyCrypto { fn new(lasthop: bool) -> Self { DummyCrypto { counter_tag: [0; 20], counter: 0, lasthop, } } } // Helper: set up a 3-hop circuit with no encryption, where the // next inbound message seems to come from hop next_msg_from async fn newcirc_ext( chan: Arc, next_msg_from: HopNum, ) -> ( Arc, reactor::Reactor, mpsc::Sender, ) { let circid = 128.into(); let (_created_send, created_recv) = oneshot::channel(); let (circmsg_send, circmsg_recv) = mpsc::channel(64); let unique_id = UniqId::new(23, 17); let (pending, mut reactor) = PendingClientCirc::new( circid, Arc::clone(&chan), created_recv, None, // circ_closed. circmsg_recv, unique_id, ); let PendingClientCirc { circ, recvcreated: _, } = pending; for idx in 0_u8..3 { let params = CircParameters::default(); let (hopf, reacf) = futures::join!( circ.add_hop( true, Box::new(DummyCrypto::new(idx == 2)), Box::new(DummyCrypto::new(idx == next_msg_from.into())), ¶ms, ), reactor.run_once() ); assert!(hopf.is_ok()); assert!(reacf.is_ok()); } (circ, reactor, circmsg_send) } // Helper: set up a 3-hop circuit with no encryption, where the // next inbound message seems to come from hop next_msg_from async fn newcirc( chan: Arc, ) -> ( Arc, reactor::Reactor, mpsc::Sender, ) { newcirc_ext(chan, 2.into()).await } // Try sending a cell via send_relay_cell #[async_test] async fn send_simple() { let (chan, mut ch) = fake_channel(); let (circ, _reactor, _send) = newcirc(chan).await; let begindir = RelayCell::new(0.into(), RelayMsg::BeginDir); circ.send_relay_cell(2.into(), false, begindir) .await .unwrap(); // Here's what we tried to put on the TLS channel. Note that // we're using dummy relay crypto for testing convenience. let rcvd = ch.cells.next().await.unwrap(); assert_eq!(rcvd.circid(), 128.into()); let m = match rcvd.into_circid_and_msg().1 { ChanMsg::Relay(r) => RelayCell::decode(r.into_relay_body()).unwrap(), _ => panic!(), }; assert!(matches!(m.msg(), RelayMsg::BeginDir)); } // Try getting a "meta-cell", which is what we're calling those not // for a specific circuit. #[async_test] async fn recv_meta() { let (chan, _ch) = fake_channel(); let (circ, mut reactor, mut sink) = newcirc(chan).await; // 1: Try doing it via handle_meta_cell directly. let meta_receiver = circ.register_meta_handler(2.into()).await.unwrap(); let extended: RelayMsg = relaymsg::Extended2::new((*b"123").into()).into(); { circ.c .lock() .await .handle_meta_cell(2.into(), extended.clone()) .await .unwrap(); } let msg = meta_receiver.await.unwrap().unwrap(); assert!(matches!(msg, RelayMsg::Extended2(_))); // 2: Try doing it via the reactor. let meta_receiver = circ.register_meta_handler(2.into()).await.unwrap(); sink.send(rmsg_to_ccmsg(0, extended.clone())).await.unwrap(); reactor.run_once().await.unwrap(); let msg = meta_receiver.await.unwrap().unwrap(); assert!(matches!(msg, RelayMsg::Extended2(_))); // 3: Try getting a meta cell that we didn't want. let e = { circ.c .lock() .await .handle_meta_cell(2.into(), extended.clone()) .await .err() .unwrap() }; assert_eq!( format!("{}", e), "circuit protocol violation: Unexpected EXTENDED2 cell on client circuit" ); // 3: Try getting a meta from a hop that we didn't want. let _receiver = circ.register_meta_handler(2.into()).await.unwrap(); let e = { circ.c .lock() .await .handle_meta_cell(1.into(), extended.clone()) .await .err() .unwrap() }; assert_eq!( format!("{}", e), "circuit protocol violation: Unexpected EXTENDED2 cell from hop 1 on client circuit" ); } #[async_test] async fn extend() { use crate::crypto::handshake::{ntor::NtorServer, ServerHandshake}; let (chan, mut ch) = fake_channel(); let (circ, mut reactor, mut sink) = newcirc(chan).await; let params = CircParameters::default(); let extend_fut = async move { let target = example_target(); let mut rng = thread_rng(); circ.extend_ntor(&mut rng, &target, ¶ms).await.unwrap(); circ // gotta keep the circ alive, or the reactor would exit. }; let reply_fut = async move { // We've disabled encryption on this circuit, so we can just // read the extend2 cell. let (id, chmsg) = ch.cells.next().await.unwrap().into_circid_and_msg(); assert_eq!(id, 128.into()); let rmsg = match chmsg { ChanMsg::RelayEarly(r) => RelayCell::decode(r.into_relay_body()).unwrap(), _ => panic!(), }; let e2 = match rmsg.msg() { RelayMsg::Extend2(e2) => e2, _ => panic!(), }; let mut rng = thread_rng(); let (_, reply) = NtorServer::server(&mut rng, &[example_ntor_key()], e2.handshake()).unwrap(); let extended2 = relaymsg::Extended2::new(reply).into(); sink.send(rmsg_to_ccmsg(0, extended2)).await.unwrap(); sink // gotta keep the sink alive, or the reactor will exit. }; let reactor_fut = async move { reactor.run_once().await.unwrap(); // to deliver the relay cell reactor.run_once().await.unwrap(); // to handle the AddHop }; let (circ, _, _) = futures::join!(extend_fut, reply_fut, reactor_fut); // Did we really add another hop? assert_eq!(circ.n_hops().await, 4); } async fn bad_extend_test_impl(reply_hop: HopNum, bad_reply: ClientCircChanMsg) -> Error { let (chan, _ch) = fake_channel(); let (circ, mut reactor, mut sink) = newcirc_ext(chan, reply_hop).await; let params = CircParameters::default(); let extend_fut = async move { let target = example_target(); let mut rng = thread_rng(); let outcome = circ.extend_ntor(&mut rng, &target, ¶ms).await; (outcome, circ) // keep the circ alive, or the reactor will exit. }; let bad_reply_fut = async move { sink.send(bad_reply).await.unwrap(); sink // keep the sink alive, or the reactor will exit. }; let reactor_fut = async move { let res = reactor.run_once().await; if res.is_err() { reactor.propagate_close().await; } }; let ((outcome, circ), _, _) = futures::join!(extend_fut, bad_reply_fut, reactor_fut); assert_eq!(circ.n_hops().await, 3); assert!(outcome.is_err()); outcome.unwrap_err() } #[async_test] async fn bad_extend_wronghop() { let extended2 = relaymsg::Extended2::new(vec![]).into(); let cc = rmsg_to_ccmsg(0, extended2); let error = bad_extend_test_impl(1.into(), cc).await; // This case shows up as a CircDestroy, since a message sent // from the wrong hop won't even be delivered to the extend // code's meta-handler. Instead the unexpected message will cause // the circuit to get torn down. match error { Error::CircDestroy(s) => { assert_eq!(s, "Circuit closed while waiting for EXTENDED2"); } _ => panic!(), } } #[async_test] async fn bad_extend_wrongtype() { let extended = relaymsg::Extended::new(vec![7; 200]).into(); let cc = rmsg_to_ccmsg(0, extended); let error = bad_extend_test_impl(2.into(), cc).await; match error { Error::CircProto(s) => { assert_eq!(s, "wanted EXTENDED2; got EXTENDED") } _ => panic!(), } } #[async_test] async fn bad_extend_destroy() { let cc = ClientCircChanMsg::Destroy(chanmsg::Destroy::new(4.into())); let error = bad_extend_test_impl(2.into(), cc).await; match error { Error::CircDestroy(s) => assert_eq!(s, "Circuit closed while waiting for EXTENDED2"), _ => panic!(), } } #[async_test] async fn bad_extend_crypto() { let extended2 = relaymsg::Extended2::new(vec![99; 256]).into(); let cc = rmsg_to_ccmsg(0, extended2); let error = bad_extend_test_impl(2.into(), cc).await; assert!(matches!(error, Error::BadHandshake)); } #[async_test] async fn begindir() { let (chan, mut ch) = fake_channel(); let (circ, mut reactor, mut sink) = newcirc(chan).await; let begin_and_send_fut = async move { // Here we'll say we've got a circuit, and we want to // make a simple BEGINDIR request with it. let mut stream = circ.begin_dir_stream().await.unwrap(); stream.write_all(b"HTTP/1.0 GET /\r\n").await.unwrap(); stream.flush().await.unwrap(); let mut buf = [0_u8; 1024]; let n = stream.read(&mut buf).await.unwrap(); assert_eq!(&buf[..n], b"HTTP/1.0 404 Not found\r\n"); let n = stream.read(&mut buf).await.unwrap(); assert_eq!(n, 0); stream }; let reply_fut = async move { // We've disabled encryption on this circuit, so we can just // read the begindir cell. let (id, chmsg) = ch.cells.next().await.unwrap().into_circid_and_msg(); assert_eq!(id, 128.into()); // hardcoded circid. let rmsg = match chmsg { ChanMsg::Relay(r) => RelayCell::decode(r.into_relay_body()).unwrap(), _ => panic!(), }; let (streamid, rmsg) = rmsg.into_streamid_and_msg(); assert!(matches!(rmsg, RelayMsg::BeginDir)); // Reply with a Connected cell to indicate success. let connected = relaymsg::Connected::new_empty().into(); sink.send(rmsg_to_ccmsg(streamid, connected)).await.unwrap(); // Now read a DATA cell... let (id, chmsg) = ch.cells.next().await.unwrap().into_circid_and_msg(); assert_eq!(id, 128.into()); let rmsg = match chmsg { ChanMsg::Relay(r) => RelayCell::decode(r.into_relay_body()).unwrap(), _ => panic!(), }; let (streamid_2, rmsg) = rmsg.into_streamid_and_msg(); assert_eq!(streamid_2, streamid); if let RelayMsg::Data(d) = rmsg { assert_eq!(d.as_ref(), &b"HTTP/1.0 GET /\r\n"[..]); } else { panic!(); } // Write another data cell in reply! let data = relaymsg::Data::new(b"HTTP/1.0 404 Not found\r\n").into(); sink.send(rmsg_to_ccmsg(streamid, data)).await.unwrap(); // Send an END cell to say that the conversation is over. let end = relaymsg::End::new_with_reason(relaymsg::EndReason::DONE).into(); sink.send(rmsg_to_ccmsg(streamid, end)).await.unwrap(); sink // gotta keep the sink alive, or the reactor will exit. }; let reactor_fut = async move { reactor.run_once().await.unwrap(); // AddStream reactor.run_once().await.unwrap(); // Register stream closer reactor.run_once().await.unwrap(); // Connected cell reactor.run_once().await.unwrap(); // Data cell reactor.run_once().await.unwrap(); // End cell reactor }; let (_stream, _, _) = futures::join!(begin_and_send_fut, reply_fut, reactor_fut); } // Set up a circuit and stream that expects some incoming SENDMEs. async fn setup_incoming_sendme_case( n_to_send: usize, ) -> ( Arc, DataStream, mpsc::Sender, StreamId, crate::circuit::reactor::Reactor, usize, ) { let (chan, mut ch) = fake_channel(); let (circ, mut reactor, mut sink) = newcirc(chan).await; let (snd_done, mut rcv_done) = oneshot::channel::<()>(); let circ_clone = Arc::clone(&circ); let begin_and_send_fut = async move { // Take our circuit and make a stream on it. let mut stream = circ_clone .begin_stream("www.example.com", 443, None) .await .unwrap(); let junk = [0_u8; 1024]; let mut remaining = n_to_send; while remaining > 0 { let n = std::cmp::min(remaining, junk.len()); stream.write_all(&junk[..n]).await.unwrap(); remaining -= n; } stream.flush().await.unwrap(); stream }; let receive_fut = async move { // Read the begindir cell. let (_id, chmsg) = ch.cells.next().await.unwrap().into_circid_and_msg(); let rmsg = match chmsg { ChanMsg::Relay(r) => RelayCell::decode(r.into_relay_body()).unwrap(), _ => panic!(), }; let (streamid, rmsg) = rmsg.into_streamid_and_msg(); assert!(matches!(rmsg, RelayMsg::Begin(_))); // Reply with a connected cell... let connected = relaymsg::Connected::new_empty().into(); sink.send(rmsg_to_ccmsg(streamid, connected)).await.unwrap(); // Now read bytes from the stream until we have them all. let mut bytes_received = 0_usize; let mut cells_received = 0_usize; while bytes_received < n_to_send { // Read a data cell, and remember how much we got. let (id, chmsg) = ch.cells.next().await.unwrap().into_circid_and_msg(); assert_eq!(id, 128.into()); let rmsg = match chmsg { ChanMsg::Relay(r) => RelayCell::decode(r.into_relay_body()).unwrap(), _ => panic!(), }; let (streamid2, rmsg) = rmsg.into_streamid_and_msg(); assert_eq!(streamid2, streamid); if let RelayMsg::Data(dat) = rmsg { cells_received += 1; bytes_received += dat.as_ref().len(); } else { panic!() } } snd_done.send(()).unwrap(); (sink, streamid, cells_received) }; let reactor_fut = async move { use futures::FutureExt; loop { futures::select! { r = reactor.run_once().fuse() => r.unwrap(), _ = rcv_done => break, } } reactor }; let (stream, (sink, streamid, cells_received), reactor) = futures::join!(begin_and_send_fut, receive_fut, reactor_fut); (circ, stream, sink, streamid, reactor, cells_received) } #[async_test] async fn accept_valid_sendme() { let (circ, _stream, mut sink, streamid, mut reactor, cells_received) = setup_incoming_sendme_case(300 * 498 + 3).await; assert_eq!(cells_received, 301); // Make sure that the circuit is indeed expecting the right sendmes { let mut c = circ.c.lock().await; let hop = c.hop_mut(2.into()).unwrap(); let (window, tags) = hop.sendwindow.window_and_expected_tags().await; assert_eq!(window, 1000 - 301); assert_eq!(tags.len(), 3); // 100 assert_eq!(tags[0], hex!("6400000000000000000000000000000000000000")); // 200 assert_eq!(tags[1], hex!("c800000000000000000000000000000000000000")); // 300 assert_eq!(tags[2], hex!("2c01000000000000000000000000000000000000")); } let reply_with_sendme_fut = async move { // make and send a circuit-level sendme. let c_sendme = relaymsg::Sendme::new_tag(hex!("6400000000000000000000000000000000000000")).into(); sink.send(rmsg_to_ccmsg(0_u16, c_sendme)).await.unwrap(); // Make and send a stream-level sendme. let s_sendme = relaymsg::Sendme::new_empty().into(); sink.send(rmsg_to_ccmsg(streamid, s_sendme)).await.unwrap(); sink }; let reactor_fut = async move { reactor.run_once().await.unwrap(); // circuit sendme reactor.run_once().await.unwrap(); // stream sendme reactor }; let (_, _) = futures::join!(reply_with_sendme_fut, reactor_fut); // Now make sure that the circuit is still happy, and its // window is updated. { let mut c = circ.c.lock().await; let hop = c.hop_mut(2.into()).unwrap(); let (window, _tags) = hop.sendwindow.window_and_expected_tags().await; assert_eq!(window, 1000 - 201); } } #[async_test] async fn invalid_circ_sendme() { // Same setup as accept_valid_sendme() test above but try giving // a sendme with the wrong tag. let (_circ, _stream, mut sink, _streamid, mut reactor, _cells_received) = setup_incoming_sendme_case(300 * 498 + 3).await; let reply_with_sendme_fut = async move { // make and send a circuit-level sendme with a bad tag. let c_sendme = relaymsg::Sendme::new_tag(hex!("FFFF0000000000000000000000000000000000FF")).into(); sink.send(rmsg_to_ccmsg(0_u16, c_sendme)).await.unwrap(); sink }; let reactor_fut = async move { use crate::util::err::ReactorError; let r = reactor.run_once().await; match r { Err(ReactorError::Err(Error::CircProto(m))) => { assert_eq!(m, "bad auth tag on circuit sendme") } _ => panic!(), } reactor }; let (_, _) = futures::join!(reply_with_sendme_fut, reactor_fut); // TODO: check that the circuit is shut down too } #[test] fn basic_params() { use super::CircParameters; let mut p = CircParameters::default(); assert_eq!(p.initial_send_window(), 1000); assert!(p.extend_by_ed25519_id()); assert!(p.set_initial_send_window(500).is_ok()); p.set_extend_by_ed25519_id(false); assert_eq!(p.initial_send_window(), 500); assert!(!p.extend_by_ed25519_id()); assert!(p.set_initial_send_window(9000).is_err()); assert_eq!(p.initial_send_window(), 500); } }