Skip to main content

tor_proto/circuit/
reactor.rs

1//! Module exposing the circuit reactor subsystem.
2//!
3//! This module implements the new [multi-reactor circuit subsystem].
4//!
5// Note: this is currently only used for the relay side,
6// but we plan to eventually rewrite client circuit implementation
7// to use these new reactor types as well.
8//!
9//! The entry point of the reactor is [`Reactor::run`], which launches the
10//! reactor background tasks, and begins listening for inbound cells on the provided
11//! inbound Tor channel.
12//!
13//! ### Architecture
14//!
15//! Internally, the circuit reactor consists of multiple reactors,
16//! each running in a separate task:
17//!
18//!   * [`StreamReactor`] (one per hop): handles all messages arriving to,
19//!     and coming from the streams of a given hop. The ready stream messages
20//!     are sent to the [`BackwardReactor`]
21//!   * [`ForwardReactor`]: handles incoming cells arriving on the
22//!     "inbound" Tor channel (towards the guard, if we are a client, or towards
23//!     the client, if we are a relay). If we are a client, it moves stream messages
24//!     towards the corresponding [`StreamReactor`]. If we are a relay,
25//!     in addition to sending any stream messages to the `StreamReactor`,
26//!     this reactor also moves cells in the forward direction
27//!     (from the client towards the exit)
28//!   * [`BackwardReactor`]: writes cells to the "inbound" Tor channel:
29//!     towards the client if we are a relay, or the towards the exit
30//!     if we are a client.
31//!
32// TODO: the forward/backward terminology no longer makes sense! Come up with better terms...
33//!
34//! If we are an exit relay, the cell flow looks roughly like this:
35//!
36//! ```text
37//!                             <stream_tx
38//!                              MPSC (0)>
39//!   +--------------> FWD -------------------------+
40//!   |                 |                           |
41//!   |                 |                           |
42//!   |                 |                           |
43//!   |                 |                           v
44//! relay      BackwardReactorCmd            StreamReactor
45//!   ^             <MPSC (0)>                      |
46//!   |                 |                           |
47//!   |                 |                           |
48//!   |                 |                           |
49//!   |                 v                           |
50//!   +--------------- BWD <------------------------+
51//!     application stream data    <stream_rx
52//!                                 MPSC (0)>
53//!
54//! For a middle relay (the `StreamReactor` is omitted for brevity,
55//! but middle relays can have one too, if leaky pipe is in use):
56//!
57//! ```text                   unrecognized cell
58//!   +--------------> FWD -------------------------+
59//!   |                 |                           |
60//!   |                 |                           |
61//!   |                 |                           |
62//!   |                 |                           v
63//! client      BackwardReactorCmd                relay
64//! or relay        <MPSC (0)>                      |
65//!   ^                 |                           |
66//!   |                 |                           |
67//!   |                 |                           |
68//!   |                 |                           |
69//!   |                 v                           |
70//!   +--------------- BWD <------------------------+
71//! ```
72//!
73//! On the client-side the `ForwardReactor` reads cells from the Tor channel to the guard,
74//! and the `BackwardReactor` writes to it.
75//!
76//! ```text
77//!   +--------------- FWD <--------------------+
78//!   |                 |                       |
79//!   |                 |                       |
80//!   |                 |                       |
81//!   v                 |                       |
82//! StreamReactor  BackwardReactorCmd         guard
83//!   |               <MPSC (0)>                ^
84//!   |                 |                       |
85//!   |                 |                       |
86//!   |                 |                       |
87//!   |                 v                       |
88//!   +--------------> BWD ---------------------+
89//! ```
90//!
91//! Client with leaky pipe (`SR` = `StreamReactor`):
92//!
93//! ```text
94//!   +------------------------------+
95//!   |       +--------------------+ | (1 MPSC TX per SR)
96//!   |       |                    | |
97//!   |       |       +----------- FWD <------------------+
98//!   |       |       |             |                     |
99//!   |       |       |             |                     |
100//!   |       |       |             |                     |
101//!   v       v       v             |                     |
102//!  SR      SR      SR           BackwardReactorCmd    guard
103//! (hop 4) (hop 3)  (hop 2)      <MPSC (0)>              ^
104//!   |       |       |             |                     |
105//!   |       |       |             |                     |
106//!   |       |       |             |                     |
107//!   |       |       |             v                     |
108//!   |       |       |            BWD -------------------+
109//!   |       |       |             ^
110//!   |       |       |             |
111//!   |       |       |             | <stream_rx
112//!   |       |       |             |  MPSC (0)>
113//!   +-------+-------+-------------+
114//! ```
115//!
116// TODO(tuning): The inter-reactor MPSC channels have no buffering,
117// which is likely going to be bad for performance,
118// so we will need to tune the sizes of these MPSC buffers.
119//!
120//! The read and write ends of the inbound and outbound Tor channels are "split",
121//! such that each reactor holds an `inbound_chan_rx` stream (for reading)
122//! and a `inbound_chan_tx` sink (for writing):
123//!
124//!  * `ForwardReactor` holds the reading end of the inbound
125//!    (coming from the client, if we are a relay, or coming from the guard, if we are a client)
126//!    Tor channel, and the writing end of the outbound (towards the exit, if we are a middle relay)
127//!    Tor channel, if there is one
128//!  * `BackwardReactor` holds the reading end of the outbound channel, if there is one,
129//!    and the writing end of the inbound channel, if there is one
130//!
131//! #### `ForwardReactor`
132//!
133//! It handles forward cells, by delegating to the implementation-dependent
134//! [`ForwardHandler::handle_forward_cell`], which decides
135//! whether the cell needs to be handled in `ForwardReactor`,
136//! or in the `ForwardHandler` itself.
137//!
138//! More concretely:
139//!
140//! ```text
141//!
142//! Legend: `F` = "forward reactor", `H` = "ForwardHandler"
143//!
144//! | Message           | Received in | Handled in | Description                            |
145//! |-------------------|-------------|------------|----------------------------------------|
146//! | DESTROY           | F           | H          | Handled internally by the FowardHandler|
147//! |-------------------|-------------|------------|----------------------------------------|
148//! | PADDING_NEGOTIATE | F           | H          | Handled internally by the FowardHandler|
149//! |-------------------|-------------|------------|----------------------------------------|
150//! | *unrecognized*    | F           | H          | Unrecognized relay cell handling is    |
151//! | RELAY OR          |             |            | implementation-dependent so these are  |
152//! | RELAY_EARLY       |             |            | handled in the ForwardHandler.         |
153//! |                   |             |            |                                        |
154//! |                   |             |            | The relay ForwardHandler will handle   |
155//! |                   |             |            | these by forwarding them to the next   |
156//! |                   |             |            | hop, if there is one.                  |
157//! |                   |             |            |                                        |
158//! |                   |             |            | Clients don't yet implement            |
159//! |                   |             |            | ForwardHandler, but when they do,      |
160//! |                   |             |            | its implementation will simply reject  |
161//! |                   |             |            | any messages that can't be decrypted   |
162//! |-------------------|-------------|------------|----------------------------------------|
163//! | *recognized*      | F           | see table  | Handling depends on the cmd            |
164//! | RELAY OR          |             | below      |                                        |
165//! | RELAY_EARLY       |             |            |                                        |
166//! ```
167//!
168//! Recognized relay cells are handled by splitting each cell into individual messages,
169//! and handling each message individually as described in the table below
170//! (Note: since prop340 is not yet implemented, in practice there is only 1 message per cell):
171//!
172//! ```text
173//!
174//! Legend: `F` = "forward reactor", `B` = "backward reactor", `S` = "stream reactor"
175//!
176//! | RELAY cmd         | Received in | Handled in | Description                            |
177//! |-------------------|-------------|------------|----------------------------------------|
178//! | SENDME            | F           | B          | Sent to BackwardReactor for handling   |
179//! |                   |             |            | (BackwardReactorCmd::HandleSendme)     |
180//! |                   |             |            | because the forward reactor doesn't    |
181//! |                   |             |            | have access to the inbound_chan_tx part|
182//! |                   |             |            | of the inbound (towards the client)    |
183//! |                   |             |            | Tor channel, and so cannot obtain the  |
184//! |                   |             |            | congestion signals needed for SENDME   |
185//! |                   |             |            | handling                               |
186//! |-------------------|-------------|------------|----------------------------------------|
187//! | Other             | F           | F          | Passed to impl-dependent handler       |
188//! | (StreamId = 0)    |             |            |  `ForwardHandler::handle_meta_msg()`   |
189//! |-------------------|-------------|------------|----------------------------------------|
190//! | Other             | F           | S          | All messages with a non-zero stream ID |
191//! | (StreamId != 0)   |             |            | are forwarded to the stream reactor    |
192//! |-------------------|-------------|------------|----------------------------------------|
193//! ```
194//!
195//! #### `BackwardReactor`
196//!
197//! It handles
198//!
199//!  * the packaging and delivery of all cells that need to be written to the "inbound" Tor channel
200//!    (it writes them to the towards-the-client Tor channel sink) (**partially implemented**)
201//!  * incoming cells coming over the "outbound" Tor channel. This channel only exists
202//!    if we are a middle relay. These cells are relayed to the "inbound" Tor channel (**not implemented**).
203//!  * the sending of padding cells, according to the PaddingController's instructions
204//!
205//! This multi-reactor architecture should, in theory, have better performance than
206//! a single reactor system, because it enables us to parallelize some of the work:
207//! the forward and backward directions share little state,
208//! because they read from, and write to, different sinks/streams,
209//! so they can be run in parallel (as separate tasks).
210//! With a single reactor architecture, the reactor would need to drive
211//! both the forward and the backward direction, and on each iteration
212//! would need to decide which to prioritize, which might prove tricky
213//! (though prioritizing one of them at random would've probably been good enough).
214//!
215//! The monolithic single reactor alternative would also have been significantly
216//! more convoluted, and so more difficult to maintain in the long run.
217//!
218//
219// NOTE: The FWD and BWD currently share the hop list containing the per-hop state,
220// (including the congestion control object, which is behind a mutex).
221//
222//! [multi-reactor circuit subsystem]: https://gitlab.torproject.org/tpo/core/arti/-/blob/main/doc/dev/notes/relay-conflux.md
223//! [`StreamReactor`]: stream::StreamReactor
224
225// TODO(DEDUP): this will replace CircHopList when we rewrite the client reactor
226// to use the new reactor architecture
227pub(crate) mod circhop;
228
229pub(crate) mod backward;
230pub(crate) mod forward;
231pub(crate) mod hop_mgr;
232pub(crate) mod macros;
233pub(crate) mod stream;
234
235use std::result::Result as StdResult;
236use std::sync::Arc;
237
238use derive_deftly::Deftly;
239use futures::channel::mpsc;
240use futures::{FutureExt as _, StreamExt as _, select_biased};
241use oneshot_fused_workaround as oneshot;
242use tracing::trace;
243
244use tor_cell::chancell::CircId;
245use tor_rtcompat::{DynTimeProvider, Runtime};
246
247use crate::channel::Channel;
248use crate::circuit::reactor::backward::BackwardHandler;
249use crate::circuit::reactor::forward::ForwardHandler;
250use crate::circuit::reactor::hop_mgr::HopMgr;
251use crate::circuit::reactor::stream::ReadyStreamMsg;
252use crate::circuit::{CircuitRxReceiver, UniqId};
253use crate::memquota::CircuitAccount;
254use crate::util::err::ReactorError;
255
256// TODO(circpad): once padding is stabilized, the padding module will be moved out of client.
257use crate::client::circuit::padding::{PaddingController, PaddingEventStream};
258
259use backward::BackwardReactor;
260use forward::ForwardReactor;
261use macros::derive_deftly_template_CircuitReactor;
262
263/// The type of a oneshot channel used to inform reactor of the result of an operation.
264pub(crate) type ReactorResultChannel<T> = oneshot::Sender<crate::Result<T>>;
265
266/// A handle for interacting with a circuit reactor.
267#[derive(derive_more::Debug)]
268pub(crate) struct CircReactorHandle<F: ForwardHandler, B: BackwardHandler> {
269    /// Sender for reactor control messages.
270    #[debug(skip)]
271    pub(crate) control: mpsc::UnboundedSender<CtrlMsg<F::CtrlMsg, B::CtrlMsg>>,
272    /// Sender for reactor control commands.
273    #[debug(skip)]
274    pub(crate) command: mpsc::UnboundedSender<CtrlCmd<F::CtrlCmd, B::CtrlCmd>>,
275    /// The time provider.
276    #[expect(unused)] // TODO(relay)
277    pub(crate) time_provider: DynTimeProvider,
278    /// Memory quota account
279    #[expect(unused)] // TODO(relay)
280    pub(crate) memquota: CircuitAccount,
281}
282
283/// A control command.
284///
285/// The difference between this and [`CtrlMsg`] is that `CtrlCmd`s
286/// never cause cells to sent on the Tor channel,
287/// while `CtrlMsg`s potentially do.
288#[allow(unused)] // TODO(relay)
289pub(crate) enum CtrlCmd<F, B> {
290    /// A control command for the forward reactor.
291    Forward(forward::CtrlCmd<F>),
292    /// A control command for the backward reactor.
293    Backward(backward::CtrlCmd<B>),
294    /// Shut down the reactor.
295    Shutdown,
296}
297
298/// A control message.
299#[allow(unused)] // TODO(relay)
300pub(crate) enum CtrlMsg<F, B> {
301    /// A control message for the forward reactor.
302    Forward(forward::CtrlMsg<F>),
303    /// A control message for the backward reactor.
304    Backward(backward::CtrlMsg<B>),
305}
306
307/// The entry point of the circuit reactor subsystem.
308#[derive(Deftly)]
309#[derive_deftly(CircuitReactor)]
310#[deftly(reactor_name = "circuit reactor")]
311#[deftly(only_run_once)]
312#[deftly(run_inner_fn = "Self::run_inner")]
313#[must_use = "If you don't call run() on a reactor, the circuit won't work."]
314pub(crate) struct Reactor<R: Runtime, F: ForwardHandler, B: BackwardHandler> {
315    /// The process-unique identifier of this circuit.
316    ///
317    /// Used for logging.
318    unique_id: UniqId,
319    /// The reactor for handling
320    ///
321    ///   * cells moving in the forward direction (from the client towards exit), if we are a relay
322    ///   * incoming cells (coming from the guard), if we are a client
323    ///
324    /// Optional so we can move it out of self in run().
325    forward: Option<ForwardReactor<R, F>>,
326    /// The reactor for handling
327    ///
328    ///   * cells moving in the backward direction (from the exit towards client), if we are a relay
329    ///   * outgoing cells (moving towards the guard), if we are a client
330    ///
331    /// Optional so we can move it out of self in run().
332    backward: Option<BackwardReactor<B>>,
333    /// Receiver for control messages for this reactor, sent by reactor handle objects.
334    control: mpsc::UnboundedReceiver<CtrlMsg<F::CtrlMsg, B::CtrlMsg>>,
335    /// Receiver for command messages for this reactor, sent by reactor handle objects.
336    ///
337    /// This MPSC channel is polled in [`run`](Self::run).
338    ///
339    /// NOTE: this is a separate channel from `control`, because some messages
340    /// have higher priority and need to be handled even if the `inbound_chan_tx` is not
341    /// ready (whereas `control` messages are not read until the `inbound_chan_tx` sink
342    /// is ready to accept cells).
343    command: mpsc::UnboundedReceiver<CtrlCmd<F::CtrlCmd, B::CtrlCmd>>,
344    /// Control channels for the [`ForwardReactor`].
345    ///
346    /// Handles [`CtrlCmd::Forward`] and [`CtrlMsg::Forward`] messages.
347    fwd_ctrl: ReactorCtrl<forward::CtrlCmd<F::CtrlCmd>, forward::CtrlMsg<F::CtrlMsg>>,
348    /// Control channels for the [`BackwardReactor`].
349    ///
350    /// Handles [`CtrlCmd::Backward`] and [`CtrlMsg::Backward`] messages.
351    bwd_ctrl: ReactorCtrl<backward::CtrlCmd<B::CtrlCmd>, backward::CtrlMsg<B::CtrlMsg>>,
352}
353
354/// A handle for sending control/command messages to a FWD or BWD.
355struct ReactorCtrl<C, M> {
356    /// Sender for control commands.
357    command_tx: mpsc::UnboundedSender<C>,
358    /// Sender for control messages.
359    control_tx: mpsc::UnboundedSender<M>,
360}
361
362impl<C, M> ReactorCtrl<C, M> {
363    /// Create a new sender handle.
364    fn new(command_tx: mpsc::UnboundedSender<C>, control_tx: mpsc::UnboundedSender<M>) -> Self {
365        Self {
366            command_tx,
367            control_tx,
368        }
369    }
370
371    /// Send a control command.
372    fn send_cmd(&mut self, cmd: C) -> Result<(), ReactorError> {
373        self.command_tx
374            .unbounded_send(cmd)
375            .map_err(|_| ReactorError::Shutdown)
376    }
377
378    /// Send a control message.
379    fn send_msg(&mut self, msg: M) -> Result<(), ReactorError> {
380        self.control_tx
381            .unbounded_send(msg)
382            .map_err(|_| ReactorError::Shutdown)
383    }
384}
385
386/// Trait implemented by types that can handle control messages and commands.
387pub(crate) trait ControlHandler {
388    /// The type of control message expected by the forward reactor.
389    type CtrlMsg;
390
391    /// The type of control command expected by the forward reactor.
392    type CtrlCmd;
393
394    // TODO(DEDUP): do these APIs make sense?
395    // What should we return here, maybe some instructions for the base reactor
396    // to do something?
397
398    /// Handle a control command.
399    fn handle_cmd(&mut self, cmd: Self::CtrlCmd) -> StdResult<(), ReactorError>;
400
401    /// Handle a control message.
402    fn handle_msg(&mut self, msg: Self::CtrlMsg) -> StdResult<(), ReactorError>;
403}
404
405#[allow(unused)] // TODO(relay)
406impl<R: Runtime, F: ForwardHandler + ControlHandler, B: BackwardHandler + ControlHandler>
407    Reactor<R, F, B>
408{
409    /// Create a new circuit reactor.
410    ///
411    /// The reactor will send outbound messages on `channel`, receive incoming
412    /// messages on `inbound_chan_rx`, and identify this circuit by the channel-local
413    /// [`CircId`] provided.
414    ///
415    /// The internal unique identifier for this circuit will be `unique_id`.
416    #[allow(clippy::too_many_arguments)] // TODO
417    pub(crate) fn new(
418        runtime: R,
419        channel: &Arc<Channel>,
420        circ_id: CircId,
421        unique_id: UniqId,
422        inbound_chan_rx: CircuitRxReceiver,
423        forward_impl: F,
424        backward_impl: B,
425        hop_mgr: HopMgr<R>,
426        padding_ctrl: PaddingController,
427        padding_event_stream: PaddingEventStream,
428        // The sending end of this channel should be in HopMgr
429        bwd_rx: mpsc::Receiver<ReadyStreamMsg>,
430        fwd_events: mpsc::Receiver<F::CircEvent>,
431        memquota: &CircuitAccount,
432    ) -> (Self, CircReactorHandle<F, B>) {
433        // NOTE: not registering this channel with the memquota subsystem is okay,
434        // because it has no buffering (if ever decide to make the size of this buffer
435        // non-zero for whatever reason, we must remember to register it with memquota
436        // so that it counts towards the total memory usage for the circuit.
437        #[allow(clippy::disallowed_methods)]
438        let (backward_reactor_tx, forward_reactor_rx) = mpsc::channel(0);
439
440        // TODO: channels galore
441        let (control_tx, control_rx) = mpsc::unbounded();
442        let (command_tx, command_rx) = mpsc::unbounded();
443
444        let (fwd_control_tx, fwd_control_rx) = mpsc::unbounded();
445        let (fwd_command_tx, fwd_command_rx) = mpsc::unbounded();
446        let (bwd_control_tx, bwd_control_rx) = mpsc::unbounded();
447        let (bwd_command_tx, bwd_command_rx) = mpsc::unbounded();
448
449        let fwd_ctrl = ReactorCtrl::new(fwd_command_tx, fwd_control_tx);
450        let bwd_ctrl = ReactorCtrl::new(bwd_command_tx, bwd_control_tx);
451
452        let handle = CircReactorHandle {
453            control: control_tx,
454            command: command_tx,
455            time_provider: DynTimeProvider::new(runtime.clone()),
456            memquota: memquota.clone(),
457        };
458
459        /// Grab a handle to the hop list (it's needed by the BWD)
460        let hops = Arc::clone(hop_mgr.hops());
461        let forward = ForwardReactor::new(
462            runtime.clone(),
463            unique_id,
464            forward_impl,
465            hop_mgr,
466            inbound_chan_rx,
467            fwd_control_rx,
468            fwd_command_rx,
469            backward_reactor_tx,
470            fwd_events,
471            padding_ctrl.clone(),
472        );
473
474        let backward = BackwardReactor::new(
475            runtime,
476            channel,
477            circ_id,
478            unique_id,
479            backward_impl,
480            hops,
481            forward_reactor_rx,
482            bwd_control_rx,
483            bwd_command_rx,
484            padding_ctrl,
485            padding_event_stream,
486            bwd_rx,
487        );
488
489        let reactor = Reactor {
490            unique_id,
491            forward: Some(forward),
492            backward: Some(backward),
493            control: control_rx,
494            command: command_rx,
495            fwd_ctrl,
496            bwd_ctrl,
497        };
498
499        (reactor, handle)
500    }
501
502    /// Helper for [`run`](Self::run).
503    pub(crate) async fn run_inner(&mut self) -> StdResult<(), ReactorError> {
504        let (forward, backward) = (|| Some((self.forward.take()?, self.backward.take()?)))()
505            .expect("relay reactor spawned twice?!");
506
507        let mut forward = Box::pin(forward.run()).fuse();
508        let mut backward = Box::pin(backward.run()).fuse();
509        loop {
510            // If either of these completes, this function returns,
511            // dropping fwd_ctrl/bwd_ctrl channels, which will, in turn,
512            // cause the remaining reactor, if there is one, to shut down too
513            select_biased! {
514                res = self.command.next() => {
515                    let Some(cmd) = res else {
516                        trace!(
517                            circ_id = %self.unique_id,
518                            reason = "command channel drop",
519                            "reactor shutdown",
520                        );
521
522                        return Err(ReactorError::Shutdown);
523                    };
524
525                    self.handle_command(cmd)?;
526                },
527                res = self.control.next() => {
528                    let Some(msg) = res else {
529                        trace!(
530                            circ_id = %self.unique_id,
531                            reason = "control channel drop",
532                            "reactor shutdown",
533                        );
534
535                        return Err(ReactorError::Shutdown);
536                    };
537
538                    self.handle_control(msg)?;
539                },
540                // No need to log the error here, because it was already logged
541                // by the reactor that shut down
542                res = forward => return Ok(res?),
543                res = backward => return Ok(res?),
544            }
545        }
546    }
547
548    /// Handle a shutdown request.
549    fn handle_shutdown(&self) -> StdResult<(), ReactorError> {
550        trace!(
551            tunnel_id = %self.unique_id,
552            "reactor shutdown due to explicit request",
553        );
554
555        Err(ReactorError::Shutdown)
556    }
557
558    /// Handle a [`CtrlCmd`].
559    fn handle_command(
560        &mut self,
561        cmd: CtrlCmd<F::CtrlCmd, B::CtrlCmd>,
562    ) -> StdResult<(), ReactorError> {
563        match cmd {
564            CtrlCmd::Forward(c) => self.fwd_ctrl.send_cmd(c),
565            CtrlCmd::Backward(c) => self.bwd_ctrl.send_cmd(c),
566            CtrlCmd::Shutdown => self.handle_shutdown(),
567        }
568    }
569
570    /// Handle a [`CtrlMsg`].
571    fn handle_control(
572        &mut self,
573        cmd: CtrlMsg<F::CtrlMsg, B::CtrlMsg>,
574    ) -> StdResult<(), ReactorError> {
575        match cmd {
576            CtrlMsg::Forward(c) => self.fwd_ctrl.send_msg(c),
577            CtrlMsg::Backward(c) => self.bwd_ctrl.send_msg(c),
578        }
579    }
580}
581
582#[cfg(test)]
583pub(crate) mod test {
584    // @@ begin test lint list maintained by maint/add_warning @@
585    #![allow(clippy::bool_assert_comparison)]
586    #![allow(clippy::clone_on_copy)]
587    #![allow(clippy::dbg_macro)]
588    #![allow(clippy::mixed_attributes_style)]
589    #![allow(clippy::print_stderr)]
590    #![allow(clippy::print_stdout)]
591    #![allow(clippy::single_char_pattern)]
592    #![allow(clippy::unwrap_used)]
593    #![allow(clippy::unchecked_time_subtraction)]
594    #![allow(clippy::useless_vec)]
595    #![allow(clippy::needless_pass_by_value)]
596    #![allow(clippy::string_slice)] // See arti#2571
597    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
598
599    use tor_basic_utils::test_rng::testing_rng;
600    use tor_cell::chancell::{BoxedCellBody, msg as chanmsg};
601    use tor_cell::relaycell::{AnyRelayMsgOuter, RelayCellFormat, StreamId, msg as relaymsg};
602
603    use chanmsg::AnyChanMsg;
604
605    #[cfg(feature = "hs-service")]
606    use crate::stream::IncomingStreamRequestFilter;
607
608    pub(crate) fn rmsg_to_ccmsg(
609        id: Option<StreamId>,
610        msg: relaymsg::AnyRelayMsg,
611        early: bool,
612    ) -> AnyChanMsg {
613        // TODO #1947: test other formats.
614        let rfmt = RelayCellFormat::V0;
615        let body: BoxedCellBody = AnyRelayMsgOuter::new(id, msg)
616            .encode(rfmt, &mut testing_rng())
617            .unwrap();
618        let chanmsg = chanmsg::Relay::from(body);
619
620        if early {
621            let chanmsg = chanmsg::RelayEarly::from(chanmsg);
622            AnyChanMsg::RelayEarly(chanmsg)
623        } else {
624            AnyChanMsg::Relay(chanmsg)
625        }
626    }
627
628    #[cfg(any(feature = "hs-service", feature = "relay"))]
629    pub(crate) struct AllowAllStreamsFilter;
630    #[cfg(any(feature = "hs-service", feature = "relay"))]
631    impl IncomingStreamRequestFilter for AllowAllStreamsFilter {
632        fn disposition(
633            &mut self,
634            _ctx: &crate::stream::IncomingStreamRequestContext<'_>,
635            _circ: &crate::circuit::CircHopSyncView<'_>,
636        ) -> crate::Result<crate::stream::IncomingStreamRequestDisposition> {
637            Ok(crate::stream::IncomingStreamRequestDisposition::Accept)
638        }
639    }
640}