diff options
Diffstat (limited to 'crates/tor-proto')
20 files changed, 248 insertions, 93 deletions
diff --git a/crates/tor-proto/semver.md b/crates/tor-proto/semver.md index 45bf8a4b0..082e5824c 100644 --- a/crates/tor-proto/semver.md +++ b/crates/tor-proto/semver.md @@ -1 +1,2 @@ BREAKING: `CreateRequestHandler::new` takes an additional argument for configuring the handling of incoming streams +BREAKING: Logging statement now use `circ_uniq_id=` and `{backward|forward}_circ_id=` for a clearer identification diff --git a/crates/tor-proto/src/channel.rs b/crates/tor-proto/src/channel.rs index 32a1e152e..ef4fe6c57 100644 --- a/crates/tor-proto/src/channel.rs +++ b/crates/tor-proto/src/channel.rs @@ -876,13 +876,13 @@ impl Channel { sender, tx, })?; - let (id, circ_unique_id, padding_ctrl, padding_stream) = + let (circ_id, circ_unique_id, padding_ctrl, padding_stream) = rx.await.map_err(|_| ChannelClosed)??; - trace!("{}: Allocated CircId {}", circ_unique_id, id); + trace!("{}: Allocated CircId {}", circ_unique_id, circ_id); Ok(PendingClientTunnel::new( - id, + circ_id, self.clone(), createdreceiver, receiver, diff --git a/crates/tor-proto/src/circuit/circhop.rs b/crates/tor-proto/src/circuit/circhop.rs index bf596e81d..5442ee840 100644 --- a/crates/tor-proto/src/circuit/circhop.rs +++ b/crates/tor-proto/src/circuit/circhop.rs @@ -26,7 +26,7 @@ use postage::watch; use safelog::sensitive as sv; use tracing::{debug, trace}; -use tor_cell::chancell::BoxedCellBody; +use tor_cell::chancell::{BoxedCellBody, CircId}; use tor_cell::relaycell::extend::{CcRequest, CircRequestExt}; use tor_cell::relaycell::flow_ctrl::{Xoff, Xon, XonKBpsEwma}; use tor_cell::relaycell::msg::AnyRelayMsg; @@ -552,13 +552,15 @@ impl CircHopOutbound { /// If no END cell is specified, an END cell with the reason byte set to /// REASON_MISC will be sent. /// - // Note(relay): `circ_id` is an opaque displayable type + // Note(relay): `circ_uniq_id` is an opaque displayable type // because relays use a different circuit ID type // than clients. Eventually, we should probably make // them both use the same ID type, or have a nicer approach here + #[allow(clippy::too_many_arguments)] pub(crate) fn close_stream( &mut self, - circ_id: impl std::fmt::Display, + circ_uniq_id: impl std::fmt::Display, + circ_id: CircId, id: StreamId, hop: Option<HopNum>, message: CloseStreamBehavior, @@ -571,6 +573,7 @@ impl CircHopOutbound { .expect("lock poisoned") .terminate(id, why, expiry)?; trace!( + circ_uniq_id = %circ_uniq_id, circ_id = %circ_id, stream_id = %id, should_send_end = ?should_send_end, @@ -681,13 +684,14 @@ impl CircHopOutbound { // // TODO prop340: This should take a cell or similar, not a message. // - // Note(relay): `circ_id` is an opaque displayable type + // Note(relay): `circ_uniq_id` is an opaque displayable type // because relays use a different circuit ID type // than clients. Eventually, we should probably make // them both use the same ID type, or have a nicer approach here pub(crate) fn about_to_send( &mut self, - circ_id: impl std::fmt::Display, + circ_uniq_id: impl std::fmt::Display, + circ_id: CircId, stream_id: StreamId, msg: &AnyRelayMsg, ) -> Result<()> { @@ -703,6 +707,7 @@ impl CircHopOutbound { // but the caller of `about_to_send()` isn't designed to handle fallible sends // so it would need some refactoring to handle this. debug!( + circ_uniq_id = %circ_uniq_id, circ_id = %circ_id, stream_id = %stream_id, "sending a relay cell for non-existent or non-open stream!", diff --git a/crates/tor-proto/src/circuit/reactor.rs b/crates/tor-proto/src/circuit/reactor.rs index 7de7a37dd..d011fafae 100644 --- a/crates/tor-proto/src/circuit/reactor.rs +++ b/crates/tor-proto/src/circuit/reactor.rs @@ -316,6 +316,8 @@ pub(crate) struct Reactor<R: Runtime, F: ForwardHandler, B: BackwardHandler> { /// /// Used for logging. unique_id: UniqId, + /// The circuit identifier on the inbound Tor channel. + circ_id: CircId, /// The reactor for handling /// /// * cells moving in the forward direction (from the client towards exit), if we are a relay @@ -461,6 +463,7 @@ impl<R: Runtime, F: ForwardHandler + ControlHandler, B: BackwardHandler + Contro let forward = ForwardReactor::new( runtime.clone(), unique_id, + circ_id, forward_impl, hop_mgr, inbound_chan_rx, @@ -488,6 +491,7 @@ impl<R: Runtime, F: ForwardHandler + ControlHandler, B: BackwardHandler + Contro let reactor = Reactor { unique_id, + circ_id, forward: Some(forward), backward: Some(backward), control: control_rx, @@ -514,7 +518,8 @@ impl<R: Runtime, F: ForwardHandler + ControlHandler, B: BackwardHandler + Contro res = self.command.next() => { let Some(cmd) = res else { trace!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, reason = "command channel drop", "reactor shutdown", ); @@ -527,7 +532,8 @@ impl<R: Runtime, F: ForwardHandler + ControlHandler, B: BackwardHandler + Contro res = self.control.next() => { let Some(msg) = res else { trace!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, reason = "control channel drop", "reactor shutdown", ); diff --git a/crates/tor-proto/src/circuit/reactor/backward.rs b/crates/tor-proto/src/circuit/reactor/backward.rs index c8acc6fec..d6e72aff9 100644 --- a/crates/tor-proto/src/circuit/reactor/backward.rs +++ b/crates/tor-proto/src/circuit/reactor/backward.rs @@ -175,7 +175,8 @@ pub(crate) trait BackwardHandler: ControlHandler { /// or a [`BackwardCellDisposition`] specifying how it should be handled. fn handle_backward_cell( &mut self, - circ_id: UniqId, + circ_uniq_id: UniqId, + circ_id: CircId, cell: Self::CircChanMsg, ) -> StdResult<BackwardCellDisposition, ReactorError>; } @@ -585,7 +586,8 @@ impl<B: BackwardHandler> BackwardReactor<B> { ForwardShutdown => { // The forward reactor has crashed, so we have to shut down. trace!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, "Backward relay reactor shutdown (forward reactor has closed)", ); @@ -633,7 +635,8 @@ impl<B: BackwardHandler> BackwardReactor<B> { self.send_relay_msg(hop, msg).await?; debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, "Extended circuit to the next hop" ); } @@ -655,7 +658,8 @@ impl<B: BackwardHandler> BackwardReactor<B> { // 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, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, hopnum=?hopnum, cmd = %cmd, "Sending backward cell" @@ -804,7 +808,10 @@ impl<B: BackwardHandler> BackwardReactor<B> { /// Handle a backward cell (moving from the exit towards the client). async fn handle_backward_cell(&mut self, cell: B::CircChanMsg) -> StdResult<(), ReactorError> { - match self.inner.handle_backward_cell(self.unique_id, cell)? { + match self + .inner + .handle_backward_cell(self.unique_id, self.circ_id, cell)? + { BackwardCellDisposition::Forward(cell) => { let cell = AnyChanCell::new(Some(self.circ_id), cell); self.inbound_chan_tx diff --git a/crates/tor-proto/src/circuit/reactor/forward.rs b/crates/tor-proto/src/circuit/reactor/forward.rs index 4125371e6..8122fbdf1 100644 --- a/crates/tor-proto/src/circuit/reactor/forward.rs +++ b/crates/tor-proto/src/circuit/reactor/forward.rs @@ -24,7 +24,7 @@ use { // TODO(circpad): once padding is stabilized, the padding module will be moved out of client. use crate::client::circuit::padding::PaddingController; -use tor_cell::chancell::msg::AnyChanMsg; +use tor_cell::chancell::{CircId, msg::AnyChanMsg}; use tor_cell::relaycell::msg::{Sendme, SendmeTag}; use tor_cell::relaycell::{ AnyRelayMsgOuter, RelayCellDecoderResult, RelayCellFormat, RelayCmd, UnparsedRelayMsg, @@ -64,6 +64,8 @@ pub(super) struct ForwardReactor<R: Runtime, F: ForwardHandler> { runtime: R, /// An identifier for logging about this reactor's circuit. unique_id: UniqId, + /// The circuit identifier on the inbound Tor channel. + circ_id: CircId, /// Implementation-dependent part of the reactor. /// /// This enables us to customize the behavior of the reactor, @@ -241,6 +243,7 @@ impl<R: Runtime, F: ForwardHandler> ForwardReactor<R, F> { pub(super) fn new( runtime: R, unique_id: UniqId, + circ_id: CircId, inner: F, hop_mgr: HopMgr<R>, inbound_chan_rx: CircuitRxReceiver, @@ -253,6 +256,7 @@ impl<R: Runtime, F: ForwardHandler> ForwardReactor<R, F> { Self { runtime, unique_id, + circ_id, inbound_chan_rx, control_rx, command_rx, @@ -295,7 +299,8 @@ impl<R: Runtime, F: ForwardHandler> ForwardReactor<R, F> { let cell = res.map_err(ReactorError::Err)?; let Some(cell) = cell else { debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, "Backward channel has closed, shutting down forward relay reactor", ); @@ -452,13 +457,15 @@ impl<R: Runtime, F: ForwardHandler> ForwardReactor<R, F> { Err(e) => { for m in msgs { debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, "Ignoring relay msg received after triggering shutdown: {m:?}", ); } if let Some(incomplete) = incomplete { debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, "Ignoring partial relay msg received after triggering shutdown: {:?}", incomplete, ); diff --git a/crates/tor-proto/src/circuit/reactor/hop_mgr.rs b/crates/tor-proto/src/circuit/reactor/hop_mgr.rs index 20522a8e2..ed929b28a 100644 --- a/crates/tor-proto/src/circuit/reactor/hop_mgr.rs +++ b/crates/tor-proto/src/circuit/reactor/hop_mgr.rs @@ -15,6 +15,7 @@ use { tor_cell::relaycell::StreamId, }; +use tor_cell::chancell::CircId; use tor_error::internal; use tor_rtcompat::Runtime; @@ -68,6 +69,8 @@ pub(crate) struct HopMgr<R: Runtime> { struct StreamReactorContext { /// An identifier for logging about this reactor's circuit. unique_id: UniqId, + /// The circuit identifier on the inbound Tor channel. + circ_id: CircId, /// The incoming stream handler. /// /// This is shared with every StreamReactor. @@ -86,6 +89,7 @@ impl<R: Runtime> HopMgr<R> { pub(crate) fn new_with_incoming_handler<S: StreamHandler>( runtime: R, unique_id: UniqId, + circ_id: CircId, handler: S, bwd_tx: mpsc::Sender<ReadyStreamMsg>, incoming_handler: IncomingStreamRequestHandler, @@ -94,6 +98,7 @@ impl<R: Runtime> HopMgr<R> { Self::new_inner( runtime, unique_id, + circ_id, handler, bwd_tx, Some(incoming_handler), @@ -108,6 +113,7 @@ impl<R: Runtime> HopMgr<R> { pub(crate) fn new<S: StreamHandler>( runtime: R, unique_id: UniqId, + circ_id: CircId, handler: S, bwd_tx: mpsc::Sender<ReadyStreamMsg>, memquota: CircuitAccount, @@ -115,6 +121,7 @@ impl<R: Runtime> HopMgr<R> { Self::new_inner( runtime, unique_id, + circ_id, handler, bwd_tx, #[cfg(any(feature = "hs-service", feature = "relay"))] @@ -127,6 +134,7 @@ impl<R: Runtime> HopMgr<R> { fn new_inner<S: StreamHandler>( runtime: R, unique_id: UniqId, + circ_id: CircId, handler: S, bwd_tx: mpsc::Sender<ReadyStreamMsg>, #[cfg(any(feature = "hs-service", feature = "relay"))] incoming_handler: Option< @@ -139,6 +147,7 @@ impl<R: Runtime> HopMgr<R> { let hops = Arc::new(RwLock::new(Default::default())); let ctx = StreamReactorContext { unique_id, + circ_id, #[cfg(any(feature = "hs-service", feature = "relay"))] incoming: Arc::new(Mutex::new(incoming_handler)), handler: Arc::new(handler), @@ -292,6 +301,7 @@ impl<R: Runtime> HopMgr<R> { hopnum, outbound, self.ctx.unique_id, + self.ctx.circ_id, fwd_stream_rx, self.bwd_tx.clone(), Arc::clone(&self.ctx.handler), diff --git a/crates/tor-proto/src/circuit/reactor/macros.rs b/crates/tor-proto/src/circuit/reactor/macros.rs index a0b6e0bce..9f80ffee6 100644 --- a/crates/tor-proto/src/circuit/reactor/macros.rs +++ b/crates/tor-proto/src/circuit/reactor/macros.rs @@ -31,7 +31,8 @@ derive_deftly::define_derive_deftly! { let unique_id = self.unique_id; tracing::debug!( - circ_id = %unique_id, + circ_uniq_id = %unique_id, + backward_circ_id = %self.circ_id, "Running {}", ${tmeta(reactor_name) as str} ); @@ -53,8 +54,17 @@ derive_deftly::define_derive_deftly! { // May log at a higher level depending on the error kind. let msg = format!("{} shut down", ${tmeta(reactor_name) as str}); match &result { - Ok(()) => tracing::trace!(circ_id = %unique_id, "{msg}"), - Err(e) => tor_error::debug_report!(e, circ_id = %unique_id, "{msg}"), + Ok(()) => tracing::trace!( + circ_uniq_id = %unique_id, + backward_circ_id = %self.circ_id, + "{msg}" + ), + Err(e) => tor_error::debug_report!( + e, + circ_uniq_id = %unique_id, + backward_circ_id = %self.circ_id, + "{msg}" + ), } result diff --git a/crates/tor-proto/src/circuit/reactor/stream.rs b/crates/tor-proto/src/circuit/reactor/stream.rs index 215f99e72..7962e9504 100644 --- a/crates/tor-proto/src/circuit/reactor/stream.rs +++ b/crates/tor-proto/src/circuit/reactor/stream.rs @@ -18,6 +18,7 @@ use crate::stream::incoming::{ }; use tor_async_utils::{SinkTrySend as _, SinkTrySendError as _}; +use tor_cell::chancell::CircId; use tor_cell::relaycell::msg::{AnyRelayMsg, Begin, BeginDir, End, EndReason, Resolve}; use tor_cell::relaycell::{ AnyRelayMsgOuter, RelayCellFormat, RelayCmd, StreamId, UnparsedRelayMsg, @@ -79,6 +80,8 @@ pub(crate) struct StreamReactor { time_provider: DynTimeProvider, /// An identifier for logging about this reactor's circuit. unique_id: UniqId, + /// The circuit identifier on the inbound Tor channel. + circ_id: CircId, /// Receiver for Tor stream data that need to be delivered to a Tor stream. /// /// The sender is in the [`HopMgr`](super::hop_mgr::HopMgr) of the @@ -120,6 +123,7 @@ impl StreamReactor { hopnum: Option<HopNum>, hop: CircHopOutbound, unique_id: UniqId, + circ_id: CircId, cell_rx: mpsc::Receiver<CtrlMsg>, bwd_tx: mpsc::Sender<ReadyStreamMsg>, inner: Arc<dyn StreamHandler>, @@ -132,6 +136,7 @@ impl StreamReactor { hop, time_provider: DynTimeProvider::new(runtime), unique_id, + circ_id, #[cfg(any(feature = "hs-service", feature = "relay"))] incoming, cell_rx, @@ -281,8 +286,12 @@ impl StreamReactor { // (the BWD handles the encoding) if c_t_w { if let Some(stream_id) = bwd_msg.stream_id() { - self.hop - .about_to_send(self.unique_id, stream_id, bwd_msg.msg())?; + self.hop.about_to_send( + self.unique_id, + self.circ_id, + stream_id, + bwd_msg.msg(), + )?; } } @@ -465,7 +474,8 @@ impl StreamReactor { // IncomingStreamRequestHandler, we need to do it elsewhere, in // a different way. debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + backward_circ_id = %self.circ_id, "Incoming stream request receiver dropped", ); // This will _cause_ the circuit to get closed. @@ -549,9 +559,15 @@ impl StreamReactor { ) -> StdResult<(), ReactorError> { let timeout = self.inner.halfstream_expiry(&self.hop); let expire_at = self.time_provider.now() + timeout; - let res = self - .hop - .close_stream(self.unique_id, sid, None, behav, reason, expire_at)?; + let res = self.hop.close_stream( + self.unique_id, + self.circ_id, + sid, + None, + behav, + reason, + expire_at, + )?; let Some(msg) = res else { // We may not need to send anything at all... return Ok(()); diff --git a/crates/tor-proto/src/client/circuit.rs b/crates/tor-proto/src/client/circuit.rs index 608d467da..a0c26933e 100644 --- a/crates/tor-proto/src/client/circuit.rs +++ b/crates/tor-proto/src/client/circuit.rs @@ -801,7 +801,7 @@ impl PendingClientTunnel { /// Does not send a CREATE* cell on its own. #[allow(clippy::too_many_arguments)] pub(crate) fn new( - id: CircId, + circ_id: CircId, channel: Arc<Channel>, createdreceiver: oneshot::Receiver<CreateResponse>, input: CircuitRxReceiver, @@ -815,7 +815,7 @@ impl PendingClientTunnel { let time_provider = channel.time_provider().clone(); let (reactor, control_tx, command_tx, reactor_closed_rx, mutable) = Reactor::new( channel, - id, + circ_id, unique_id, input, runtime, @@ -832,7 +832,7 @@ impl PendingClientTunnel { command: command_tx, reactor_closed_rx: reactor_closed_rx.shared(), #[cfg(test)] - circid: id, + circid: circ_id, memquota, time_provider, is_multi_path: false, diff --git a/crates/tor-proto/src/client/reactor.rs b/crates/tor-proto/src/client/reactor.rs index fd8435be9..ab74aa8fd 100644 --- a/crates/tor-proto/src/client/reactor.rs +++ b/crates/tor-proto/src/client/reactor.rs @@ -637,7 +637,7 @@ impl Reactor { #[allow(clippy::type_complexity, clippy::too_many_arguments)] // TODO pub(super) fn new( channel: Arc<Channel>, - channel_id: CircId, + circ_id: CircId, unique_id: UniqId, input: CircuitRxReceiver, runtime: DynTimeProvider, @@ -669,7 +669,7 @@ impl Reactor { let circuit_leg = Circuit::new( runtime.clone(), channel, - channel_id, + circ_id, unique_id, input, memquota, diff --git a/crates/tor-proto/src/client/reactor/circuit.rs b/crates/tor-proto/src/client/reactor/circuit.rs index cc32ef4e3..72127db3a 100644 --- a/crates/tor-proto/src/client/reactor/circuit.rs +++ b/crates/tor-proto/src/client/reactor/circuit.rs @@ -115,8 +115,8 @@ pub(crate) struct Circuit { /// Mutable information about this circuit, /// shared with the reactor's `ConfluxSet`. mutable: Arc<MutableState>, - /// This circuit's identifier on the upstream channel. - channel_id: CircId, + /// This circuit's identifier. + circ_id: CircId, /// An identifier for logging about this reactor's circuit. unique_id: TunnelScopedCircId, /// A handler for conflux cells. @@ -220,7 +220,7 @@ impl Circuit { pub(super) fn new( runtime: DynTimeProvider, channel: Arc<Channel>, - channel_id: CircId, + circ_id: CircId, unique_id: TunnelScopedCircId, input: CircuitRxReceiver, memquota: CircuitAccount, @@ -240,7 +240,7 @@ impl Circuit { crypto_in: InboundClientCrypt::new(), hops: CircHopList::default(), unique_id, - channel_id, + circ_id, crypto_out, mutable, #[cfg(feature = "conflux")] @@ -259,6 +259,11 @@ impl Circuit { self.unique_id.unique_id() } + /// Return this circuit's identifier. + pub(super) fn circ_id(&self) -> CircId { + self.circ_id + } + /// Return the shared mutable state of this circuit. pub(super) fn mutable(&self) -> &Arc<MutableState> { &self.mutable @@ -431,7 +436,12 @@ impl Circuit { return Err(internal!("tried to send cell on unlinked circuit").into()); } - trace!(circ_id = %self.unique_id, cell = ?msg, "sending relay cell"); + trace!( + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, + cell = ?msg, + "sending relay cell" + ); // Cloned, because we borrow mutably from self when we get the circhop. let runtime = self.runtime.clone(); @@ -504,14 +514,20 @@ impl Circuit { leg: UniqId, cell: ClientCircChanMsg, ) -> Result<Vec<CircuitCmd>> { - trace!(circ_id = %self.unique_id, cell = ?cell, "handling cell"); + trace!( + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, + cell = ?cell, + "handling cell" + ); use ClientCircChanMsg::*; match cell { Relay(r) => self.handle_relay_cell(handlers, leg, r), Destroy(d) => { let reason = d.reason(); debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, "Received DESTROY cell. Reason: {} [{}]", reason.human_str(), reason @@ -964,7 +980,8 @@ impl Circuit { // IncomingStreamRequestHandler, we need to do it elsewhere, in // a different way. debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, "Incoming stream request receiver dropped", ); // This will _cause_ the circuit to get closed. @@ -1046,7 +1063,8 @@ impl Circuit { let (state, msg) = H::client1(&mut rand::rng(), key, msg)?; let create_cell = wrap.to_chanmsg(msg); trace!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, create = %create_cell.cmd(), "Extending to hop 1", ); @@ -1065,7 +1083,11 @@ impl Circuit { .relay_crypt_protocol() .construct_client_layers(HandshakeRole::Initiator, keygen)?; - trace!(circ_id = %self.unique_id, "Handshake complete; circuit created."); + trace!( + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, + "Handshake complete; circuit created." + ); let peer_id = self.channel.target().clone(); @@ -1180,7 +1202,7 @@ impl Circuit { let hop_num = (hop_num as u8).into(); - let hop = CircHop::new(self.unique_id, hop_num, settings); + let hop = CircHop::new(self.unique_id, self.circ_id, hop_num, settings); self.hops.push(hop); self.crypto_in.add_layer(rev); self.crypto_out.add_layer(fwd); @@ -1237,7 +1259,8 @@ impl Circuit { .into_msg(); let reason = truncated.reason(); debug!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, "Truncated from hop {}. Reason: {} [{}]", hopnum.display(), reason.human_str(), @@ -1258,7 +1281,12 @@ impl Circuit { } } - trace!(circ_id = %self.unique_id, cell = ?msg, "Received meta-cell"); + trace!( + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, + cell = ?msg, + "Received meta-cell" + ); #[cfg(feature = "conflux")] if matches!( @@ -1274,7 +1302,8 @@ impl Circuit { if self.is_conflux_pending() { warn!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, "received unexpected cell {msg:?} on unlinked conflux circuit", ); return Err(Error::CircProto( @@ -1295,7 +1324,8 @@ impl Circuit { // Somebody was waiting for a message -- maybe this message let ret = handler.handle_msg(msg, self); trace!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, result = ?ret, "meta handler completed", ); @@ -1371,7 +1401,7 @@ impl Circuit { msg: AnyChanMsg, info: Option<QueuedCellPaddingInfo>, ) -> Result<()> { - let cell = AnyChanCell::new(Some(self.channel_id), msg); + let cell = AnyChanCell::new(Some(self.circ_id), msg); // Note: this future is always `Ready`, so await won't block. Pin::new(&mut self.chan_sender) .send_unbounded((cell, info)) @@ -1629,6 +1659,6 @@ impl Circuit { impl Drop for Circuit { fn drop(&mut self) { - let _ = self.channel.close_circuit(self.channel_id); + let _ = self.channel.close_circuit(self.circ_id); } } diff --git a/crates/tor-proto/src/client/reactor/circuit/circhop.rs b/crates/tor-proto/src/client/reactor/circuit/circhop.rs index c862ab756..c2e27fe1d 100644 --- a/crates/tor-proto/src/client/reactor/circuit/circhop.rs +++ b/crates/tor-proto/src/client/reactor/circuit/circhop.rs @@ -17,7 +17,7 @@ use crate::{Error, Result}; use futures::Stream; use futures::stream::FuturesUnordered; use smallvec::SmallVec; -use tor_cell::chancell::BoxedCellBody; +use tor_cell::chancell::{BoxedCellBody, CircId}; use tor_cell::relaycell::flow_ctrl::{Xoff, Xon, XonKBpsEwma}; use tor_cell::relaycell::msg::AnyRelayMsg; use tor_cell::relaycell::{ @@ -211,6 +211,8 @@ impl CircHopList { pub(crate) struct CircHop { /// The unique ID of the circuit. Used for logging. unique_id: TunnelScopedCircId, + /// The Tor circuit identifier. Used for logging. + circ_id: CircId, /// Hop number in the path. hop_num: HopNum, /// The inbound state of the hop. @@ -227,6 +229,7 @@ impl CircHop { /// Create a new hop. pub(crate) fn new( unique_id: TunnelScopedCircId, + circ_id: CircId, hop_num: HopNum, settings: &HopSettings, ) -> Self { @@ -244,6 +247,7 @@ impl CircHop { CircHop { unique_id, + circ_id, hop_num, inbound, outbound, @@ -279,8 +283,15 @@ impl CircHop { why: streammap::TerminateReason, expiry: Instant, ) -> Result<Option<SendRelayCell>> { - self.outbound - .close_stream(self.unique_id, id, Some(self.hop_num), message, why, expiry) + self.outbound.close_stream( + self.unique_id, + self.circ_id, + id, + Some(self.hop_num), + message, + why, + expiry, + ) } /// Check if we should send an XON message. @@ -332,7 +343,8 @@ impl CircHop { // // TODO prop340: This should take a cell or similar, not a message. pub(crate) fn about_to_send(&mut self, stream_id: StreamId, msg: &AnyRelayMsg) -> Result<()> { - self.outbound.about_to_send(self.unique_id, stream_id, msg) + self.outbound + .about_to_send(self.unique_id, self.circ_id, stream_id, msg) } /// Add an entry to this map using the specified StreamId. diff --git a/crates/tor-proto/src/client/reactor/circuit/extender.rs b/crates/tor-proto/src/client/reactor/circuit/extender.rs index 3a2520ace..d7ace65cd 100644 --- a/crates/tor-proto/src/client/reactor/circuit/extender.rs +++ b/crates/tor-proto/src/client/reactor/circuit/extender.rs @@ -12,6 +12,7 @@ use crate::{Error, Result}; use crate::{HopLocation, congestion}; use oneshot_fused_workaround as oneshot; use std::borrow::Borrow; +use tor_cell::chancell::CircId; use tor_cell::chancell::msg::HandshakeType; use tor_cell::relaycell::msg::{Extend2, Extended2}; use tor_cell::relaycell::{AnyRelayMsgOuter, UnparsedRelayMsg}; @@ -44,6 +45,8 @@ where settings: HopSettings, /// An identifier for logging about this reactor's circuit. unique_id: TunnelScopedCircId, + /// The circuit identifier on the channel. + circ_id: CircId, /// The hop we're expecting the EXTENDED2 cell to come back from. expected_hop: HopNum, /// A oneshot channel that we should inform when we are done with this extend operation. @@ -77,12 +80,14 @@ where match (|| { let mut rng = rand::rng(); let unique_id = circ.unique_id; + let circ_id = circ.circ_id; let (state, msg) = H::client1(&mut rng, key, client_aux_data)?; let n_hops = circ.crypto_out.n_layers(); let hop = ((n_hops - 1) as u8).into(); trace!( - circ_id = %unique_id, + circ_uniq_id = %unique_id, + forward_circ_id = %circ_id, target_hop = n_hops + 1, linkspecs = ?linkspecs, "Extending circuit", @@ -96,13 +101,18 @@ where cell, }; - trace!(circ_id = %unique_id, "waiting for EXTENDED2 cell"); + trace!( + circ_uniq_id = %unique_id, + forward_circ_id = %circ_id, + "waiting for EXTENDED2 cell" + ); // ... and now we wait for a response. let extender = Self { peer_id, state: Some(state), settings, unique_id, + circ_id, expected_hop: hop, operation_finished: None, }; @@ -137,7 +147,8 @@ where let relay_handshake = msg.into_body(); trace!( - circ_id = %self.unique_id, + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, "Received EXTENDED2 cell; completing handshake.", ); // Now perform the second part of the handshake, and see if it @@ -158,8 +169,12 @@ where .relay_crypt_protocol() .construct_client_layers(HandshakeRole::Initiator, keygen)?; - trace!(circ_id = %self.unique_id, settings = ?self.settings, - "Handshake complete; circuit extended."); + trace!( + circ_uniq_id = %self.unique_id, + forward_circ_id = %self.circ_id, + settings = ?self.settings, + "Handshake complete; circuit extended." + ); // If we get here, it succeeded. Add a new hop to the circuit. circ.add_hop( diff --git a/crates/tor-proto/src/client/reactor/conflux.rs b/crates/tor-proto/src/client/reactor/conflux.rs index 76bd4ecab..1d6993ff5 100644 --- a/crates/tor-proto/src/client/reactor/conflux.rs +++ b/crates/tor-proto/src/client/reactor/conflux.rs @@ -296,7 +296,8 @@ impl ConfluxSet { let circ = self.remove_unchecked(leg)?; tracing::trace!( - circ_id = %circ.unique_id(), + circ_uniq_id = %circ.unique_id(), + forward_circ_id = %circ.circ_id(), "Circuit removed from conflux set" ); @@ -942,6 +943,7 @@ impl ConfluxSet { for leg in &mut self.legs { let unique_id = leg.unique_id(); + let circ_id = leg.circ_id(); let tunnel_id = self.tunnel_id; let runtime = runtime.clone(); @@ -1011,7 +1013,11 @@ impl ConfluxSet { match ready_streams.next().await { Some(x) => x, None => { - info!(circ_id=%unique_id, "no ready streams (maybe blocked on cc?)"); + info!( + circ_uniq_id = %unique_id, + forward_circ_id = %circ_id, + "no ready streams (maybe blocked on cc?)" + ); // There are no ready streams (for example, they may all be // blocked due to congestion control), so there is nothing // to do. @@ -1055,7 +1061,8 @@ impl ConfluxSet { () = conflux_hs_timeout.fuse() => { warn!( tunnel_id = %tunnel_id, - circ_id = %unique_id, + circ_uniq_id = %unique_id, + forward_circ_id = %circ_id, "Conflux handshake timed out on circuit" ); @@ -1213,12 +1220,12 @@ impl ConfluxSet { /// if the removal of the leg ought to trigger a reactor shutdown. /// /// Returns an error if the leg doesn't exit in the conflux set. - fn remove_unchecked(&mut self, circ_id: UniqId) -> Result<Circuit, Bug> { + fn remove_unchecked(&mut self, circ_uniq_id: UniqId) -> Result<Circuit, Bug> { let idx = self .legs .iter() - .position(|circ| circ.unique_id() == circ_id) - .ok_or_else(|| internal!("leg {circ_id:?} not found in conflux set"))?; + .position(|circ| circ.unique_id() == circ_uniq_id) + .ok_or_else(|| internal!("leg {circ_uniq_id:?} not found in conflux set"))?; Ok(self.legs.remove(idx)) } @@ -1227,11 +1234,11 @@ impl ConfluxSet { #[cfg(feature = "circ-padding")] pub(super) async fn run_padding_event( &mut self, - circ_id: UniqId, + circ_uniq_id: UniqId, padding_event: PaddingEvent, ) -> crate::Result<()> { use PaddingEvent as E; - let Some(circ) = self.leg_mut(circ_id) else { + let Some(circ) = self.leg_mut(circ_uniq_id) else { // No such circuit; it must have gone away after generating this event. // Just ignore it. return Ok(()); 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" ); diff --git a/crates/tor-proto/src/tunnel.rs b/crates/tor-proto/src/tunnel.rs index 96fa41ebc..1f1088dd4 100644 --- a/crates/tor-proto/src/tunnel.rs +++ b/crates/tor-proto/src/tunnel.rs @@ -33,22 +33,25 @@ impl TunnelId { /// process-unique, but in the logs it's often useful to display the /// owning tunnel's ID alongside the circuit identifier. #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, Display)] -#[display("{} ({})", circ_id, tunnel_id)] +#[display("{} ({})", circ_uniq_id, tunnel_id)] pub(crate) struct TunnelScopedCircId { /// The identifier of the owning tunnel tunnel_id: TunnelId, /// The process-unique identifier of the circuit - circ_id: UniqId, + circ_uniq_id: UniqId, } impl TunnelScopedCircId { /// Create a new [`TunnelScopedCircId`] from the specified identifiers. - pub(crate) fn new(tunnel_id: TunnelId, circ_id: UniqId) -> Self { - Self { tunnel_id, circ_id } + pub(crate) fn new(tunnel_id: TunnelId, circ_uniq_id: UniqId) -> Self { + Self { + tunnel_id, + circ_uniq_id, + } } /// Return the [`UniqId`]. pub(crate) fn unique_id(&self) -> UniqId { - self.circ_id + self.circ_uniq_id } } |