diff options
Diffstat (limited to 'crates/tor-proto/src/relay')
| -rw-r--r-- | crates/tor-proto/src/relay/reactor.rs | 6 | ||||
| -rw-r--r-- | crates/tor-proto/src/relay/reactor/backward.rs | 8 | ||||
| -rw-r--r-- | crates/tor-proto/src/relay/reactor/forward.rs | 34 | ||||
| -rw-r--r-- | crates/tor-proto/src/relay/reactor/forward/extend_handler.rs | 20 |
4 files changed, 47 insertions, 21 deletions
diff --git a/crates/tor-proto/src/relay/reactor.rs b/crates/tor-proto/src/relay/reactor.rs index b8b473e3b..7b5e0507c 100644 --- a/crates/tor-proto/src/relay/reactor.rs +++ b/crates/tor-proto/src/relay/reactor.rs @@ -214,6 +214,7 @@ impl<R: Runtime> Reactor<R> { let mut hop_mgr = HopMgr::new_with_incoming_handler( runtime.clone(), unique_id, + circ_id, StreamHandler, stream_tx, incoming_handler, @@ -232,6 +233,7 @@ impl<R: Runtime> Reactor<R> { let (fwd_ev_tx, fwd_ev_rx) = mpsc::channel(0); let forward = Forward::new( channel, + circ_id, unique_id, crypto_out, chan_provider, @@ -754,7 +756,7 @@ pub(crate) mod test { // The reactor handled the EXTEND2 and launched an outbound channel assert!(logs_contain( - "Launched channel to the next hop circ_id=Circ 8.17" + "Launched channel to the next hop circ_uniq_id=Circ 8.17" )); assert!(ctrl.outbound_chan_launched()); assert!(!ctrl.is_closing()); @@ -1019,7 +1021,7 @@ pub(crate) mod test { // ... but the exit stream is not assert!(logs_contain("stream reactor shut down")); assert!(logs_contain( - "Stream protocol violation: Unexpected BEGIN on incoming stream circ_id=Circ 8.17" + "Stream protocol violation: Unexpected BEGIN on incoming stream circ_uniq_id=Circ 8.17" )); // The reactor won't create an IncomingStream, diff --git a/crates/tor-proto/src/relay/reactor/backward.rs b/crates/tor-proto/src/relay/reactor/backward.rs index 6b25cc5f2..d9119e395 100644 --- a/crates/tor-proto/src/relay/reactor/backward.rs +++ b/crates/tor-proto/src/relay/reactor/backward.rs @@ -9,7 +9,7 @@ use crate::util::err::ReactorError; use crate::{Error, HopNum}; use tor_cell::chancell::msg::{AnyChanMsg, Relay}; -use tor_cell::chancell::{BoxedCellBody, ChanCmd}; +use tor_cell::chancell::{BoxedCellBody, ChanCmd, CircId}; use tor_cell::relaycell::msg::SendmeTag; use std::result::Result as StdResult; @@ -51,7 +51,8 @@ impl BackwardHandler for Backward { fn handle_backward_cell( &mut self, - circ_id: UniqId, + circ_uniq_id: UniqId, + circ_id: CircId, cell: RelayCircChanMsg, ) -> StdResult<BackwardCellDisposition, ReactorError> { let disp = match cell { @@ -73,7 +74,8 @@ impl BackwardHandler for Backward { } RelayCircChanMsg::Destroy(d) => { debug!( - circ_id = %circ_id, + circ_uniq_id = %circ_uniq_id, + backward_circ_id = %circ_id, reason = %d.reason(), "Received inbound DESTROY, circuit shutting down", ); diff --git a/crates/tor-proto/src/relay/reactor/forward.rs b/crates/tor-proto/src/relay/reactor/forward.rs index 235a251f1..b4bfa87d2 100644 --- a/crates/tor-proto/src/relay/reactor/forward.rs +++ b/crates/tor-proto/src/relay/reactor/forward.rs @@ -54,6 +54,8 @@ const MAX_RELAY_EARLY_CELLS_PER_CIRCUIT: usize = 8; pub(crate) struct Forward { /// An identifier for logging about this reactor's circuit. unique_id: UniqId, + /// The circuit identifier on the inbound Tor channel. + circ_id: CircId, /// The outbound view of this circuit, if we are not the last hop. /// /// Delivers cells towards the exit. @@ -112,6 +114,7 @@ impl Forward { /// Create a new [`Forward`]. pub(crate) fn new( inbound_chan: &Arc<Channel>, + circ_id: CircId, unique_id: UniqId, crypto_out: Box<dyn OutboundRelayLayer + Send>, chan_provider: Arc<dyn ChannelProvider<BuildSpec = OwnedChanTarget> + Send + Sync>, @@ -119,11 +122,18 @@ impl Forward { memquota: CircuitAccount, ) -> Self { let inbound_peer = Arc::clone(inbound_chan.peer_info()); - let extend_handler = - ExtendRequestHandler::new(unique_id, chan_provider, inbound_peer, event_tx, memquota); + let extend_handler = ExtendRequestHandler::new( + unique_id, + circ_id, + chan_provider, + inbound_peer, + event_tx, + memquota, + ); Self { unique_id, + circ_id, // Initially, we are the last hop in the circuit. outbound: None, crypto_out, @@ -231,14 +241,6 @@ impl Forward { info: Option<QueuedCellPaddingInfo>, early: bool, ) -> StdResult<(), ReactorError> { - // TODO(relay): remove this log once we add some tests - // and confirm relaying cells works as expected - // (in practice it will be too noisy to be useful, even at trace level). - trace!( - circ_id = %self.unique_id, - "Forwarding unrecognized cell" - ); - let Some(chan) = self.outbound.as_mut() else { // The client shouldn't try to send us any cells before it gets // an EXTENDED2 cell from us @@ -248,6 +250,15 @@ impl Forward { .into()); }; + // TODO(relay): remove this log once we add some tests + // and confirm relaying cells works as expected + // (in practice it will be too noisy to be useful, even at trace level). + trace!( + circ_uniq_id = %self.unique_id, + forward_circ_id = %chan.circ_id, + "Forwarding unrecognized cell" + ); + let msg = Relay::from(BoxedCellBody::from(body)); let relay = if early { AnyChanMsg::RelayEarly(msg.into()) @@ -276,7 +287,8 @@ impl Forward { /// Handle a DESTROY cell originating from the client. fn handle_destroy_cell(&mut self, cell: &Destroy) -> StdResult<(), ReactorError> { debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, reason = %cell.reason(), "Received outbound DESTROY, circuit shutting down", ); diff --git a/crates/tor-proto/src/relay/reactor/forward/extend_handler.rs b/crates/tor-proto/src/relay/reactor/forward/extend_handler.rs index fab989e23..2c26b31ce 100644 --- a/crates/tor-proto/src/relay/reactor/forward/extend_handler.rs +++ b/crates/tor-proto/src/relay/reactor/forward/extend_handler.rs @@ -9,7 +9,7 @@ use crate::peer::PeerInfo; use crate::relay::channel_provider::{ChannelProvider, ChannelResult, OutboundChanSender}; use crate::relay::reactor::CircuitAccount; use crate::util::err::ReactorError; -use tor_cell::chancell::AnyChanCell; +use tor_cell::chancell::{AnyChanCell, CircId}; use tor_cell::relaycell::UnparsedRelayMsg; use tor_cell::relaycell::msg::{Extend2, Extended2}; use tor_error::{internal, into_internal, warn_report}; @@ -28,6 +28,8 @@ use std::sync::Arc; pub(super) struct ExtendRequestHandler { /// An identifier for logging about this handler. unique_id: UniqId, + /// The circuit identifier on the inbound Tor channel. + circ_id: CircId, /// Whether we have received an EXTEND2 on this circuit. /// // TODO(relay): bools can be finicky. @@ -54,6 +56,7 @@ impl ExtendRequestHandler { /// Create a new [`ExtendRequestHandler`]. pub(super) fn new( unique_id: UniqId, + circ_id: CircId, chan_provider: Arc<dyn ChannelProvider<BuildSpec = OwnedChanTarget> + Send + Sync>, inbound_peer: Arc<PeerInfo>, event_tx: mpsc::Sender<CircEvent>, @@ -61,6 +64,7 @@ impl ExtendRequestHandler { ) -> Self { Self { unique_id, + circ_id, have_seen_extend2: false, chan_provider, inbound_peer, @@ -129,6 +133,7 @@ impl ExtendRequestHandler { let mut result_tx = self.event_tx.clone(); let rt = runtime.clone(); let unique_id = self.unique_id; + let circ_id = self.circ_id; let memquota = self.memquota.clone(); // TODO(relay): because we dispatch this the entire EXTEND2 handling to a background task, @@ -137,7 +142,8 @@ impl ExtendRequestHandler { // because it runs in another task). Maybe we need to rethink the ChannelProvider API? runtime .spawn(async move { - let res = Self::extend_circuit(rt, unique_id, extend2, chan_rx, memquota).await; + let res = + Self::extend_circuit(rt, unique_id, circ_id, extend2, chan_rx, memquota).await; // Discard the error if the reactor shut down before we had // a chance to complete the extend handshake @@ -155,6 +161,7 @@ impl ExtendRequestHandler { async fn extend_circuit<R: Runtime>( _runtime: R, unique_id: UniqId, + inbound_circ_id: CircId, extend2: Extend2, mut chan_rx: mpsc::UnboundedReceiver<ChannelResult>, memquota: CircuitAccount, @@ -177,7 +184,8 @@ impl ExtendRequestHandler { }; debug!( - circ_id = %unique_id, + circ_uniq_id = %unique_id, + backward_circ_id = %inbound_circ_id, "Launched channel to the next hop" ); @@ -203,7 +211,8 @@ impl ExtendRequestHandler { let cell = AnyChanCell::new(Some(circ_id), create2); trace!( - circ_id = %unique_id, + circ_uniq_id = %unique_id, + forward_circ_id = %circ_id, "Sending CREATE2 to the next hop" ); @@ -219,7 +228,8 @@ impl ExtendRequestHandler { .map_err(|_| internal!("channel disappeared?"))?; trace!( - circ_id = %unique_id, + circ_uniq_id = %unique_id, + forward_circ_id = %circ_id, "Got CREATED2 response from next hop" ); |