diff options
Diffstat (limited to 'crates/tor-proto/src/client')
| -rw-r--r-- | crates/tor-proto/src/client/circuit.rs | 6 | ||||
| -rw-r--r-- | crates/tor-proto/src/client/reactor.rs | 4 | ||||
| -rw-r--r-- | crates/tor-proto/src/client/reactor/circuit.rs | 64 | ||||
| -rw-r--r-- | crates/tor-proto/src/client/reactor/circuit/circhop.rs | 20 | ||||
| -rw-r--r-- | crates/tor-proto/src/client/reactor/circuit/extender.rs | 25 | ||||
| -rw-r--r-- | crates/tor-proto/src/client/reactor/conflux.rs | 23 |
6 files changed, 103 insertions, 39 deletions
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(()); |