summaryrefslogtreecommitdiff
path: root/crates/tor-proto/src/circuit
diff options
context:
space:
mode:
Diffstat (limited to 'crates/tor-proto/src/circuit')
-rw-r--r--crates/tor-proto/src/circuit/circhop.rs15
-rw-r--r--crates/tor-proto/src/circuit/reactor.rs10
-rw-r--r--crates/tor-proto/src/circuit/reactor/backward.rs17
-rw-r--r--crates/tor-proto/src/circuit/reactor/forward.rs15
-rw-r--r--crates/tor-proto/src/circuit/reactor/hop_mgr.rs10
-rw-r--r--crates/tor-proto/src/circuit/reactor/macros.rs16
-rw-r--r--crates/tor-proto/src/circuit/reactor/stream.rs28
7 files changed, 86 insertions, 25 deletions
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(());
generated by cgit v1.2.3 (git 2.46.0) at 2026-08-28 19:52:17 +0000