aboutsummaryrefslogtreecommitdiff
path: root/crates
diff options
context:
space:
mode:
Diffstat (limited to 'crates')
-rw-r--r--crates/tor-proto/src/circuit/circhop.rs28
-rw-r--r--crates/tor-proto/src/circuit/reactor/stream.rs14
-rw-r--r--crates/tor-proto/src/client/reactor/circuit.rs14
-rw-r--r--crates/tor-proto/src/client/reactor/circuit/circhop.rs11
-rw-r--r--crates/tor-proto/src/relay/reactor.rs6
-rw-r--r--crates/tor-proto/src/stream/flow_ctrl/state.rs17
-rw-r--r--crates/tor-proto/src/stream/flow_ctrl/xon_xoff/reader.rs12
-rw-r--r--crates/tor-proto/src/stream/flow_ctrl/xon_xoff/state.rs12
8 files changed, 87 insertions, 27 deletions
diff --git a/crates/tor-proto/src/circuit/circhop.rs b/crates/tor-proto/src/circuit/circhop.rs
index 5442ee840..e2b9aab01 100644
--- a/crates/tor-proto/src/circuit/circhop.rs
+++ b/crates/tor-proto/src/circuit/circhop.rs
@@ -12,7 +12,9 @@ use crate::stream::SEND_WINDOW_INIT;
use crate::stream::StreamMpscSender;
use crate::stream::cmdcheck::{AnyCmdChecker, StreamStatus};
use crate::stream::flow_ctrl::params::FlowCtrlParameters;
-use crate::stream::flow_ctrl::state::{FlowCtrlHooks, StreamFlowCtrl, StreamRateLimit};
+use crate::stream::flow_ctrl::state::{
+ FlowCtrlHooks, StreamFlowCtrl, StreamRateLimit, WithSidechannelMitigations,
+};
use crate::stream::flow_ctrl::xon_xoff::reader::DrainRateRequest;
use crate::stream::queue::{StreamQueueReceiver, stream_queue};
use crate::streammap::{
@@ -509,7 +511,13 @@ impl CircHopOutbound {
let mut drain_rate_request_tx = NotifySender::new_typed();
let drain_rate_request_rx = drain_rate_request_tx.subscribe();
- let flow_ctrl = self.build_flow_ctrl(rate_limit_tx, drain_rate_request_tx)?;
+ let flow_ctrl = self.build_flow_ctrl(
+ // We are starting the stream,
+ // so we're a client and want flow control sidechannel mitigations.
+ WithSidechannelMitigations::Enabled,
+ rate_limit_tx,
+ drain_rate_request_tx,
+ )?;
let stream_queue_max_len = flow_ctrl.inbound_queue_max_len();
@@ -725,6 +733,7 @@ impl CircHopOutbound {
time_prov: &DynTimeProvider,
stream_id: StreamId,
cmd_checker: AnyCmdChecker,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
memquota: &StreamAccount,
) -> Result<ReactorStreamComponents> {
// TODO: This has a lot of duplicated code with `Self::begin_stream()`.
@@ -738,7 +747,11 @@ impl CircHopOutbound {
let mut drain_rate_request_tx = NotifySender::new_typed();
let drain_rate_request_rx = drain_rate_request_tx.subscribe();
- let flow_ctrl = self.build_flow_ctrl(rate_limit_tx, drain_rate_request_tx)?;
+ let flow_ctrl = self.build_flow_ctrl(
+ with_sidechannel_mitigations,
+ rate_limit_tx,
+ drain_rate_request_tx,
+ )?;
let stream_queue_max_len = flow_ctrl.inbound_queue_max_len();
@@ -765,6 +778,7 @@ impl CircHopOutbound {
#[cfg_attr(feature = "flowctl-cc", expect(clippy::unnecessary_wraps))]
fn build_flow_ctrl(
&self,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
rate_limit_updater: watch::Sender<StreamRateLimit>,
drain_rate_requester: NotifySender<DrainRateRequest>,
) -> Result<StreamFlowCtrl> {
@@ -781,15 +795,9 @@ impl CircHopOutbound {
} else {
cfg_if::cfg_if! {
if #[cfg(feature = "flowctl-cc")] {
- // TODO: Currently arti only supports clients, and we don't support connecting
- // to onion services while using congestion control, so we hardcode this. In the
- // future we will need to somehow tell the `CircHop` this so that we can set it
- // correctly, since we don't want to enable this at exits.
- let use_sidechannel_mitigations = true;
-
Ok(StreamFlowCtrl::new_xon_xoff(
params,
- use_sidechannel_mitigations,
+ with_sidechannel_mitigations,
rate_limit_updater,
drain_rate_requester,
))
diff --git a/crates/tor-proto/src/circuit/reactor/stream.rs b/crates/tor-proto/src/circuit/reactor/stream.rs
index 7962e9504..062500967 100644
--- a/crates/tor-proto/src/circuit/reactor/stream.rs
+++ b/crates/tor-proto/src/circuit/reactor/stream.rs
@@ -7,6 +7,7 @@ use crate::congestion::{CongestionControl, sendme};
use crate::memquota::{CircuitAccount, SpecificAccount as _, StreamAccount};
use crate::stream::CloseStreamBehavior;
use crate::stream::cmdcheck::StreamStatus;
+use crate::stream::flow_ctrl::state::WithSidechannelMitigations;
use crate::streammap;
use crate::util::err::ReactorError;
use crate::{Error, HopNum};
@@ -50,6 +51,9 @@ pub(crate) trait StreamHandler: Send + Sync + 'static {
/// This is the amount of time we are willing to wait for
/// an END ack before removing the half-stream from the map.
fn halfstream_expiry(&self, hop: &CircHopOutbound) -> Duration;
+
+ /// Whether sidechannel mitigations should be enabled for incoming streams.
+ fn flowctrl_sidechannel_mitigations(&self) -> WithSidechannelMitigations;
}
/// The stream reactor for a given hop.
@@ -436,9 +440,13 @@ impl StreamReactor {
StreamAccount::new(&self.memquota).map_err(|e| ReactorError::Err(e.into()))?;
let cmd_checker = InboundDataCmdChecker::new_connected();
- let stream_components =
- self.hop
- .add_ent_with_id(&self.time_provider, sid, cmd_checker, &memquota)?;
+ let stream_components = self.hop.add_ent_with_id(
+ &self.time_provider,
+ sid,
+ cmd_checker,
+ self.inner.flowctrl_sidechannel_mitigations(),
+ &memquota,
+ )?;
let outcome = Pin::new(&mut handler.incoming_sender).try_send(StreamReqInfo {
req,
diff --git a/crates/tor-proto/src/client/reactor/circuit.rs b/crates/tor-proto/src/client/reactor/circuit.rs
index d49c500db..4d56524e5 100644
--- a/crates/tor-proto/src/client/reactor/circuit.rs
+++ b/crates/tor-proto/src/client/reactor/circuit.rs
@@ -29,6 +29,7 @@ use crate::crypto::handshake::ntor_v3::{NtorV3Client, NtorV3PublicKey};
use crate::crypto::handshake::{ClientHandshake, KeyGenerator};
use crate::memquota::{CircuitAccount, SpecificAccount as _, StreamAccount};
use crate::stream::cmdcheck::{AnyCmdChecker, StreamStatus};
+use crate::stream::flow_ctrl::state::WithSidechannelMitigations;
use crate::stream::msg_streamid;
use crate::streammap;
use crate::tunnel::TunnelScopedCircId;
@@ -746,7 +747,16 @@ impl Circuit {
if let Some(msg) = res {
cfg_if::cfg_if! {
if #[cfg(feature = "hs-service")] {
- return self.handle_incoming_stream_request(handlers, msg, streamid, hopnum, leg);
+ return self.handle_incoming_stream_request(
+ handlers,
+ msg,
+ streamid,
+ hopnum,
+ leg,
+ // This is an onion service stream,
+ // so we want sidechannel mitigations for flow control.
+ WithSidechannelMitigations::Enabled,
+ );
} else {
return Err(
Error::CircProto(format!("Cannot handle {} cells on this circuit", msg.cmd())),
@@ -846,6 +856,7 @@ impl Circuit {
stream_id: StreamId,
hop_num: HopNum,
leg: UniqId,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
) -> Result<Option<CircuitCmd>> {
use tor_cell::relaycell::msg::EndReason;
use tor_error::into_internal;
@@ -937,6 +948,7 @@ impl Circuit {
self.chan_sender.time_provider(),
stream_id,
cmd_checker,
+ with_sidechannel_mitigations,
&memquota,
)?;
diff --git a/crates/tor-proto/src/client/reactor/circuit/circhop.rs b/crates/tor-proto/src/client/reactor/circuit/circhop.rs
index c2e27fe1d..0fd30eda9 100644
--- a/crates/tor-proto/src/client/reactor/circuit/circhop.rs
+++ b/crates/tor-proto/src/client/reactor/circuit/circhop.rs
@@ -9,6 +9,7 @@ use crate::congestion::CongestionControl;
use crate::crypto::cell::HopNum;
use crate::memquota::StreamAccount;
use crate::stream::cmdcheck::AnyCmdChecker;
+use crate::stream::flow_ctrl::state::WithSidechannelMitigations;
use crate::streammap::{self, StreamEntMut, StreamMap};
use crate::tunnel::TunnelScopedCircId;
use crate::util::tunnel_activity::TunnelActivity;
@@ -354,10 +355,16 @@ impl CircHop {
time_prov: &DynTimeProvider,
stream_id: StreamId,
cmd_checker: AnyCmdChecker,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
memquota: &StreamAccount,
) -> Result<ReactorStreamComponents> {
- self.outbound
- .add_ent_with_id(time_prov, stream_id, cmd_checker, memquota)
+ self.outbound.add_ent_with_id(
+ time_prov,
+ stream_id,
+ cmd_checker,
+ with_sidechannel_mitigations,
+ memquota,
+ )
}
/// Note that we received an END message (or other message indicating the end of
diff --git a/crates/tor-proto/src/relay/reactor.rs b/crates/tor-proto/src/relay/reactor.rs
index d7f3f19f0..80cba018d 100644
--- a/crates/tor-proto/src/relay/reactor.rs
+++ b/crates/tor-proto/src/relay/reactor.rs
@@ -69,6 +69,7 @@ use crate::relay::RelayCirc;
use crate::relay::channel_provider::ChannelProvider;
use crate::relay::reactor::backward::Backward;
use crate::relay::reactor::forward::Forward;
+use crate::stream::flow_ctrl::state::WithSidechannelMitigations;
use crate::stream::flow_ctrl::xon_xoff::reader::XonXoffReaderCtrl;
use crate::stream::incoming::{
IncomingCmdChecker, IncomingStream, IncomingStreamRequestFilter, IncomingStreamRequestHandler,
@@ -112,6 +113,11 @@ impl stream::StreamHandler for StreamHandler {
// if we don't have any RTT measurements yet
.unwrap_or_default()
}
+
+ fn flowctrl_sidechannel_mitigations(&self) -> WithSidechannelMitigations {
+ // We're a relay, so we don't want sidechannel mitigations for flow control.
+ WithSidechannelMitigations::Disabled
+ }
}
#[allow(unused)] // TODO(relay)
diff --git a/crates/tor-proto/src/stream/flow_ctrl/state.rs b/crates/tor-proto/src/stream/flow_ctrl/state.rs
index 41682fac2..ff51abeba 100644
--- a/crates/tor-proto/src/stream/flow_ctrl/state.rs
+++ b/crates/tor-proto/src/stream/flow_ctrl/state.rs
@@ -52,14 +52,14 @@ impl StreamFlowCtrl {
#[cfg(feature = "flowctl-cc")]
pub(crate) fn new_xon_xoff(
params: Arc<FlowCtrlParameters>,
- use_sidechannel_mitigations: bool,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
rate_limit_updater: watch::Sender<StreamRateLimit>,
drain_rate_requester: NotifySender<DrainRateRequest>,
) -> Self {
Self {
inner: StreamFlowCtrlInner::XonXoff(XonXoffFlowCtrl::new(
params,
- use_sidechannel_mitigations,
+ with_sidechannel_mitigations,
rate_limit_updater,
drain_rate_requester,
)),
@@ -261,6 +261,19 @@ impl StreamRateLimit {
}
}
+/// Whether sidechannel mitigations are enabled or not for flow control.
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+pub(crate) enum WithSidechannelMitigations {
+ /// Flow control sidechannel mitigations are *enabled*.
+ ///
+ /// Should be enabled for clients (including onion services).
+ Enabled,
+ /// Flow control sidechannel mitigations are *disabled*.
+ ///
+ /// Should be disabled for exits.
+ Disabled,
+}
+
impl std::fmt::Display for StreamRateLimit {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{} bytes/s", self.rate)
diff --git a/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/reader.rs b/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/reader.rs
index 2d02c0d18..182a0fef3 100644
--- a/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/reader.rs
+++ b/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/reader.rs
@@ -204,7 +204,9 @@ mod test {
use std::sync::atomic::{AtomicU64, Ordering};
use crate::stream::flow_ctrl::params::FlowCtrlParameters;
- use crate::stream::flow_ctrl::state::{FlowCtrlHooks, StreamRateLimit};
+ use crate::stream::flow_ctrl::state::{
+ FlowCtrlHooks, StreamRateLimit, WithSidechannelMitigations,
+ };
use crate::stream::flow_ctrl::xon_xoff::state::XonXoffFlowCtrl;
use crate::util::notify::NotifySender;
@@ -345,7 +347,7 @@ mod test {
/// 4. The flow control logic.
#[allow(clippy::type_complexity)]
fn init_flow_ctrl(
- use_sidechannel_mitigations: bool,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
) -> (
WriterWithLength<Compat<DuplexStream>>,
XonXoffReader<ReaderWithLength<Compat<DuplexStream>>, TestingDrainRateUpdates>,
@@ -365,7 +367,7 @@ mod test {
// The flow control logic.
let flow_ctrl = XonXoffFlowCtrl::new(
Arc::new(params),
- use_sidechannel_mitigations,
+ with_sidechannel_mitigations,
rate_limit_tx,
drain_rate_request_tx,
);
@@ -443,7 +445,7 @@ mod test {
// This is the stream queue for incoming data.
// So the `reader` is the stream reader and the `writer` would be within the reactor.
let (mut writer, mut reader, mut drain_rate_receiver, mut flow_ctrl) =
- init_flow_ctrl(/* use_sidechannel_mitigations= */ true);
+ init_flow_ctrl(WithSidechannelMitigations::Enabled);
// Data has arrived on the stream.
// We always consider sending an XOFF when a stream has received data.
@@ -528,7 +530,7 @@ mod test {
// This is the stream queue for incoming data.
// So the `reader` is the stream reader and the `writer` would be within the reactor.
let (mut writer, mut reader, mut drain_rate_receiver, mut flow_ctrl) =
- init_flow_ctrl(/* use_sidechannel_mitigations= */ true);
+ init_flow_ctrl(WithSidechannelMitigations::Enabled);
// Data has arrived on the stream.
// We always consider sending an XOFF when a stream has received data.
diff --git a/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/state.rs b/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/state.rs
index 49795cc68..b2236edc1 100644
--- a/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/state.rs
+++ b/crates/tor-proto/src/stream/flow_ctrl/xon_xoff/state.rs
@@ -43,7 +43,9 @@ use tracing::trace;
use super::reader::DrainRateRequest;
use crate::stream::flow_ctrl::params::{CellCount, FlowCtrlParameters};
-use crate::stream::flow_ctrl::state::{FlowCtrlHooks, HalfStreamFlowCtrlHooks, StreamRateLimit};
+use crate::stream::flow_ctrl::state::{
+ FlowCtrlHooks, HalfStreamFlowCtrlHooks, StreamRateLimit, WithSidechannelMitigations,
+};
use crate::util::notify::NotifySender;
use crate::{Error, Result};
@@ -83,12 +85,14 @@ impl XonXoffFlowCtrl {
/// Returns a new xon/xoff-based state.
pub(crate) fn new(
params: Arc<FlowCtrlParameters>,
- use_sidechannel_mitigations: bool,
+ with_sidechannel_mitigations: WithSidechannelMitigations,
rate_limit_updater: watch::Sender<StreamRateLimit>,
drain_rate_requester: NotifySender<DrainRateRequest>,
) -> Self {
- let sidechannel_mitigation =
- use_sidechannel_mitigations.then_some(SidechannelMitigation::new());
+ let sidechannel_mitigation = match with_sidechannel_mitigations {
+ WithSidechannelMitigations::Enabled => Some(SidechannelMitigation::new()),
+ WithSidechannelMitigations::Disabled => None,
+ };
// We use the same XOFF limit regardless of if we're a client or exit.
// See https://gitlab.torproject.org/tpo/core/torspec/-/issues/371#note_3260658
generated by cgit v1.2.3 (git 2.46.0) at 2026-08-26 17:17:22 +0000