tor_proto/client/reactor/control.rs
1//! Module providing [`CtrlMsg`].
2
3use super::circuit::extender::CircuitExtender;
4use super::{
5 CircuitHandshake, CloseStreamBehavior, MetaCellHandler, Reactor, ReactorResultChannel,
6 RunOnceCmdInner, SendRelayCell,
7};
8use crate::Result;
9use crate::circuit::celltypes::CreateResponse;
10use crate::circuit::circhop::{HopSettings, ReactorStreamComponents};
11#[cfg(feature = "circ-padding-manual")]
12use crate::client::circuit::padding;
13use crate::client::circuit::path;
14use crate::client::reactor::{NoJoinPointError, NtorClient, ReactorError};
15use crate::client::{HopLocation, TargetHop};
16use crate::crypto::binding::CircuitBinding;
17use crate::crypto::cell::{InboundClientLayer, OutboundClientLayer};
18use crate::crypto::handshake::ntor_v3::{NtorV3Client, NtorV3PublicKey};
19use crate::memquota::StreamAccount;
20use crate::stream::cmdcheck::AnyCmdChecker;
21use crate::streammap;
22use crate::util::skew::ClockSkew;
23use crate::util::tunnel_activity::TunnelActivity;
24#[cfg(test)]
25use crate::{circuit::UniqId, client::circuit::CircParameters, crypto::cell::HopNum};
26use tor_cell::chancell::msg::HandshakeType;
27use tor_cell::relaycell::flow_ctrl::XonKbpsEwma;
28use tor_cell::relaycell::msg::{AnyRelayMsg, Sendme};
29use tor_cell::relaycell::{AnyRelayMsgOuter, RelayCellFormat, StreamId};
30use tor_error::{Bug, bad_api_usage, internal, into_bad_api_usage};
31use tracing::{debug, trace};
32#[cfg(feature = "hs-service")]
33use {
34 crate::client::reactor::IncomingStreamRequestHandler,
35 crate::stream::IncomingStreamRequestFilter, crate::stream::incoming::StreamReqSender,
36};
37
38#[cfg(test)]
39use tor_cell::relaycell::msg::SendmeTag;
40
41#[cfg(feature = "conflux")]
42use super::{Circuit, ConfluxLinkResultChannel};
43
44use oneshot_fused_workaround as oneshot;
45
46use crate::crypto::handshake::ntor::NtorPublicKey;
47use tor_linkspec::{EncodedLinkSpec, OwnedChanTarget};
48
49use std::result::Result as StdResult;
50
51/// A message telling the reactor to do something.
52///
53/// For each `CtrlMsg`, the reactor will send a cell on the underlying channel.
54///
55/// The difference between this and [`CtrlCmd`] is that `CtrlMsg`s
56/// cause the reactor to send cells on the reactor's `chan_sender`,
57/// whereas `CtrlCmd` do not.
58#[derive(educe::Educe)]
59#[educe(Debug)]
60pub(crate) enum CtrlMsg {
61 /// Create the first hop of this circuit.
62 Create {
63 /// A oneshot channel on which we'll receive the creation response.
64 recv_created: oneshot::Receiver<CreateResponse>,
65 /// The handshake type to use for the first hop.
66 handshake: CircuitHandshake,
67 /// Other parameters relevant for circuit creation.
68 settings: HopSettings,
69 /// Oneshot channel to notify on completion.
70 done: ReactorResultChannel<()>,
71 },
72 /// Extend a circuit by one hop, using the ntor handshake.
73 ExtendNtor {
74 /// The peer that we're extending to.
75 ///
76 /// Used to extend our record of the circuit's path.
77 peer_id: OwnedChanTarget,
78 /// The handshake type to use for this hop.
79 public_key: NtorPublicKey,
80 /// Information about how to connect to the relay we're extending to.
81 linkspecs: Vec<EncodedLinkSpec>,
82 /// Other parameters we are negotiating.
83 settings: HopSettings,
84 /// Oneshot channel to notify on completion.
85 done: ReactorResultChannel<()>,
86 },
87 /// Extend a circuit by one hop, using the ntorv3 handshake.
88 ExtendNtorV3 {
89 /// The peer that we're extending to.
90 ///
91 /// Used to extend our record of the circuit's path.
92 peer_id: OwnedChanTarget,
93 /// The handshake type to use for this hop.
94 public_key: NtorV3PublicKey,
95 /// Information about how to connect to the relay we're extending to.
96 linkspecs: Vec<EncodedLinkSpec>,
97 /// Other parameters we are negotiating.
98 settings: HopSettings,
99 /// Oneshot channel to notify on completion.
100 done: ReactorResultChannel<()>,
101 },
102 /// Begin a stream with the provided hop in this circuit.
103 ///
104 /// Allocates a stream ID, and sends the provided message to that hop.
105 BeginStream {
106 /// The hop number to begin the stream with.
107 hop: TargetHop,
108 /// The message to send.
109 message: AnyRelayMsg,
110 /// The stream account to use for anything we allocate for the purpose of this stream.
111 memquota: StreamAccount,
112 /// Oneshot channel to notify on completion, with the allocated stream ID.
113 done: ReactorResultChannel<(
114 StreamId,
115 HopLocation,
116 RelayCellFormat,
117 ReactorStreamComponents,
118 )>,
119 /// A `CmdChecker` to keep track of which message types are acceptable.
120 cmd_checker: AnyCmdChecker,
121 },
122 /// Close the specified pending incoming stream, sending the provided END message.
123 ///
124 /// A stream is said to be pending if the message for initiating the stream was received but
125 /// not has not been responded to yet.
126 ///
127 /// This should be used by responders for closing pending incoming streams initiated by the
128 /// other party on the circuit.
129 #[cfg(feature = "hs-service")]
130 ClosePendingStream {
131 /// The hop number the stream is on.
132 hop: HopLocation,
133 /// The stream ID to send the END for.
134 stream_id: StreamId,
135 /// The END message to send, if any.
136 message: CloseStreamBehavior,
137 /// Oneshot channel to notify on completion.
138 done: ReactorResultChannel<()>,
139 },
140 /// Send a given control message on this circuit.
141 #[cfg(feature = "send-control-msg")]
142 SendMsg {
143 /// The hop to receive this message.
144 hop: TargetHop,
145 /// The message to send.
146 msg: AnyRelayMsg,
147 /// A sender that we use to tell the caller that the message was sent
148 /// and the handler installed.
149 sender: oneshot::Sender<Result<()>>,
150 },
151 /// Send a given control message on this circuit, and install a control-message handler to
152 /// receive responses.
153 #[cfg(feature = "send-control-msg")]
154 SendMsgAndInstallHandler {
155 /// The message to send, if any
156 msg: Option<AnyRelayMsgOuter>,
157 /// A message handler to install.
158 ///
159 /// If this is `None`, there must already be a message handler installed
160 #[educe(Debug(ignore))]
161 handler: Option<Box<dyn MetaCellHandler + Send + 'static>>,
162 /// A sender that we use to tell the caller that the message was sent
163 /// and the handler installed.
164 sender: oneshot::Sender<Result<()>>,
165 },
166 /// Inform the reactor that there's a flow control update for a given stream.
167 ///
168 /// The reactor will decide how to handle this update depending on the type of flow control and
169 /// the current state of the stream.
170 FlowCtrlUpdate {
171 /// The type of flow control update, and any associated metadata.
172 msg: FlowCtrlMsg,
173 /// The stream ID that the update is for.
174 stream_id: StreamId,
175 /// The hop that the stream is on.
176 hop: HopLocation,
177 },
178 /// Get the clock skew claimed by the first hop of the circuit.
179 FirstHopClockSkew {
180 /// Oneshot channel to return the clock skew.
181 answer: oneshot::Sender<StdResult<ClockSkew, Bug>>,
182 },
183 /// Link the specified circuits into the current tunnel,
184 /// to form a multi-path tunnel.
185 #[cfg(feature = "conflux")]
186 #[allow(unused)] // TODO(conflux)
187 LinkCircuits {
188 /// The circuits to link into the tunnel,
189 #[educe(Debug(ignore))]
190 circuits: Vec<Circuit>,
191 /// Oneshot channel to notify sender when all the specified circuits have finished linking,
192 /// or have failed to link.
193 ///
194 /// A client circuit is said to be fully linked once the `RELAY_CONFLUX_LINKED_ACK` is sent
195 /// (see [set construction]).
196 ///
197 /// [set construction]: https://spec.torproject.org/proposals/329-traffic-splitting.html#set-construction
198 answer: ConfluxLinkResultChannel,
199 },
200}
201
202/// A message telling the reactor to do something.
203///
204/// The difference between this and [`CtrlMsg`] is that `CtrlCmd`s
205/// never cause cells to sent on the channel,
206/// while `CtrlMsg`s potentially do: `CtrlMsg`s are mapped to [`RunOnceCmdInner`] commands,
207/// some of which instruct the reactor to send cells down the channel.
208#[derive(educe::Educe)]
209#[educe(Debug)]
210pub(crate) enum CtrlCmd {
211 /// Shut down the reactor.
212 Shutdown,
213 /// Extend the circuit by one hop, in response to an out-of-band handshake.
214 ///
215 /// (This is used for onion services, where the negotiation takes place in
216 /// INTRODUCE and RENDEZVOUS messages.)
217 #[cfg(feature = "hs-common")]
218 ExtendVirtual {
219 /// The cryptographic algorithms and keys to use when communicating with
220 /// the newly added hop.
221 #[educe(Debug(ignore))]
222 cell_crypto: (
223 Box<dyn OutboundClientLayer + Send>,
224 Box<dyn InboundClientLayer + Send>,
225 Option<CircuitBinding>,
226 ),
227 /// A set of parameters to negotiate with this hop.
228 settings: HopSettings,
229 /// Oneshot channel to notify on completion.
230 done: ReactorResultChannel<()>,
231 },
232 /// Resolve a given [`TargetHop`] into a precise [`HopLocation`].
233 ResolveTargetHop {
234 /// The target hop to resolve.
235 hop: TargetHop,
236 /// Oneshot channel to notify on completion.
237 done: ReactorResultChannel<HopLocation>,
238 },
239 /// Begin accepting streams on this circuit.
240 #[cfg(feature = "hs-service")]
241 AwaitStreamRequest {
242 /// A channel for sending information about an incoming stream request.
243 incoming_sender: StreamReqSender,
244 /// A `CmdChecker` to keep track of which message types are acceptable.
245 cmd_checker: AnyCmdChecker,
246 /// Oneshot channel to notify on completion.
247 done: ReactorResultChannel<()>,
248 /// The hop that is allowed to create streams.
249 hop: TargetHop,
250 /// A filter used to check requests before passing them on.
251 #[educe(Debug(ignore))]
252 #[cfg(feature = "hs-service")]
253 filter: Box<dyn IncomingStreamRequestFilter>,
254 },
255 /// Request the binding key of a target hop.
256 #[cfg(feature = "hs-service")]
257 GetBindingKey {
258 /// The hop for which we want the key.
259 hop: TargetHop,
260 /// Oneshot channel to notify on completion.
261 done: ReactorResultChannel<Option<CircuitBinding>>,
262 },
263 /// (tests only) Add a hop to the list of hops on this circuit, with dummy cryptography.
264 #[cfg(test)]
265 AddFakeHop {
266 relay_cell_format: RelayCellFormat,
267 fwd_lasthop: bool,
268 rev_lasthop: bool,
269 peer_id: path::HopDetail,
270 params: CircParameters,
271 done: ReactorResultChannel<()>,
272 },
273 /// (tests only) Get the send window and expected tags for a given hop.
274 #[cfg(test)]
275 QuerySendWindow {
276 hop: HopNum,
277 leg: UniqId,
278 done: ReactorResultChannel<(u32, Vec<SendmeTag>)>,
279 },
280 /// Shut down the reactor, and return the underlying [`Circuit`],
281 /// if the tunnel is not multi-path.
282 ///
283 /// Returns an error if called on a multi-path reactor.
284 #[cfg(feature = "conflux")]
285 #[allow(unused)] // TODO(conflux)
286 ShutdownAndReturnCircuit {
287 /// Oneshot channel to return the underlying [`Circuit`],
288 /// or an error if the reactor's tunnel is multi-path.
289 answer: oneshot::Sender<StdResult<Circuit, Bug>>,
290 },
291
292 /// Install or remove a [`padding::CircuitPadder`] for a given hop.
293 ///
294 /// Any existing `CircuitPadder` at that hop is replaced.
295 #[cfg(feature = "circ-padding-manual")]
296 SetPadder {
297 /// The hop to modify.
298 hop: HopLocation,
299 /// The Padder to install, or None to remove any existing padder.
300 padder: Option<padding::CircuitPadder>,
301 /// A sender to alert after we've changed the padding.
302 sender: oneshot::Sender<Result<()>>,
303 },
304
305 /// Yield the most active [`TunnelActivity`] for any hop on any leg of this tunnel.
306 GetTunnelActivity {
307 /// A sender to receive the reply.
308 sender: oneshot::Sender<TunnelActivity>,
309 },
310}
311
312/// A flow control update message.
313#[derive(Debug)]
314pub(crate) enum FlowCtrlMsg {
315 /// Send a SENDME message on this stream.
316 Sendme,
317 /// Send an XON message on this stream with the given rate.
318 Xon(XonKbpsEwma),
319}
320
321/// A control message handler object. Keep a reference to the Reactor tying its lifetime to it.
322///
323/// Its `handle_msg` and `handle_cmd` handlers decide how messages and commands,
324/// respectively, are handled.
325pub(crate) struct ControlHandler<'a> {
326 /// Reference to the reactor of this
327 reactor: &'a mut Reactor,
328}
329
330impl<'a> ControlHandler<'a> {
331 /// Constructor.
332 pub(crate) fn new(reactor: &'a mut Reactor) -> Self {
333 Self { reactor }
334 }
335
336 /// Handle a control message.
337 pub(super) fn handle_msg(&mut self, msg: CtrlMsg) -> Result<Option<RunOnceCmdInner>> {
338 trace!(
339 tunnel_id = %self.reactor.tunnel_id,
340 msg = ?msg,
341 "reactor received control message"
342 );
343
344 match msg {
345 // This is handled earlier, since it requires blocking.
346 CtrlMsg::Create { done, .. } => {
347 if self.reactor.circuits.len() == 1 {
348 // This should've been handled in Reactor::run_once()
349 // (ControlHandler::handle_msg() is never called before wait_for_create()).
350 debug_assert!(self.reactor.circuits.single_leg()?.has_hops());
351 // Don't care if the receiver goes away
352 let _ = done.send(Err(tor_error::bad_api_usage!(
353 "cannot create first hop twice"
354 )
355 .into()));
356 } else {
357 // Don't care if the receiver goes away
358 let _ = done.send(Err(tor_error::bad_api_usage!(
359 "cannot create first hop on multipath tunnel"
360 )
361 .into()));
362 }
363
364 Ok(None)
365 }
366 CtrlMsg::ExtendNtor {
367 peer_id,
368 public_key,
369 linkspecs,
370 settings,
371 done,
372 } => {
373 let Ok(circ) = self.reactor.circuits.single_leg_mut() else {
374 // Don't care if the receiver goes away
375 let _ = done.send(Err(tor_error::bad_api_usage!(
376 "cannot extend multipath tunnel"
377 )
378 .into()));
379
380 return Ok(None);
381 };
382
383 let (extender, cell) = CircuitExtender::<NtorClient>::begin(
384 peer_id,
385 HandshakeType::NTOR,
386 &public_key,
387 linkspecs,
388 settings,
389 &(),
390 circ,
391 done,
392 )?;
393 self.reactor
394 .cell_handlers
395 .set_meta_handler(Box::new(extender))?;
396
397 Ok(Some(RunOnceCmdInner::Send {
398 leg: circ.unique_id(),
399 cell,
400 done: None,
401 }))
402 }
403 CtrlMsg::ExtendNtorV3 {
404 peer_id,
405 public_key,
406 linkspecs,
407 settings,
408 done,
409 } => {
410 let Ok(circ) = self.reactor.circuits.single_leg_mut() else {
411 // Don't care if the receiver goes away
412 let _ = done.send(Err(tor_error::bad_api_usage!(
413 "cannot extend multipath tunnel"
414 )
415 .into()));
416
417 return Ok(None);
418 };
419
420 let client_extensions = settings.circuit_request_extensions()?;
421
422 let (extender, cell) = CircuitExtender::<NtorV3Client>::begin(
423 peer_id,
424 HandshakeType::NTOR_V3,
425 &public_key,
426 linkspecs,
427 settings,
428 &client_extensions,
429 circ,
430 done,
431 )?;
432 self.reactor
433 .cell_handlers
434 .set_meta_handler(Box::new(extender))?;
435
436 Ok(Some(RunOnceCmdInner::Send {
437 leg: circ.unique_id(),
438 cell,
439 done: None,
440 }))
441 }
442 CtrlMsg::BeginStream {
443 hop,
444 message,
445 memquota,
446 done,
447 cmd_checker,
448 } => {
449 // If resolving the hop fails,
450 // we want to report an error back to the initiator and not shut down the reactor.
451 let hop_location = match self.reactor.resolve_target_hop(hop) {
452 Ok(x) => x,
453 Err(e) => {
454 let e = into_bad_api_usage!("Could not resolve {hop:?}")(e);
455 // don't care if receiver goes away
456 let _ = done.send(Err(e.into()));
457 return Ok(None);
458 }
459 };
460 let (leg_id, hop_num) = match self.reactor.resolve_hop_location(hop_location) {
461 Ok(x) => x,
462 Err(e) => {
463 let e = into_bad_api_usage!("Could not resolve {hop_location:?}")(e);
464 // don't care if receiver goes away
465 let _ = done.send(Err(e.into()));
466 return Ok(None);
467 }
468 };
469 let circ = match self.reactor.circuits.leg_mut(leg_id) {
470 Some(x) => x,
471 None => {
472 let e = bad_api_usage!("Circuit leg {leg_id:?} does not exist");
473 // don't care if receiver goes away
474 let _ = done.send(Err(e.into()));
475 return Ok(None);
476 }
477 };
478
479 let result = circ.begin_stream(
480 hop_num,
481 message,
482 &self.reactor.runtime,
483 cmd_checker,
484 &memquota,
485 );
486
487 let (cell, stream_id, stream_components) = match result {
488 Ok((cell, stream_id, receiver)) => (cell, stream_id, receiver),
489 Err(e) => {
490 // don't care if receiver goes away.
491 let _ = done.send(Err(e.clone()));
492 return Err(e);
493 }
494 };
495
496 Ok(Some(RunOnceCmdInner::BeginStream {
497 cell,
498 stream_id,
499 hop: hop_location,
500 leg: leg_id,
501 stream_components,
502 done,
503 }))
504 }
505 #[cfg(feature = "hs-service")]
506 CtrlMsg::ClosePendingStream {
507 hop,
508 stream_id,
509 message,
510 done,
511 } => Ok(Some(RunOnceCmdInner::CloseStream {
512 hop,
513 sid: stream_id,
514 behav: message,
515 reason: streammap::TerminateReason::ExplicitEnd,
516 done: Some(done),
517 })),
518 CtrlMsg::FlowCtrlUpdate {
519 msg,
520 stream_id,
521 hop,
522 } => {
523 match msg {
524 FlowCtrlMsg::Sendme => {
525 let (leg_id, hop_num) = match self.reactor.resolve_hop_location(hop) {
526 Ok(x) => x,
527 Err(NoJoinPointError) => {
528 // A stream tried to send a stream-level SENDME message to the join point of
529 // a tunnel that has never had a join point. Currently in arti, only a
530 // `StreamTarget` asks us to send a stream-level SENDME, and this tunnel
531 // originally created the `StreamTarget` to begin with. So this is a
532 // legitimate bug somewhere in the tunnel code.
533 return Err(
534 internal!(
535 "Could not send a stream-level SENDME to a join point on a tunnel without a join point",
536 )
537 .into()
538 );
539 }
540 };
541
542 // Congestion control decides if we can send stream level SENDMEs or not.
543 let sendme_required = match self.reactor.uses_stream_sendme(leg_id, hop_num)
544 {
545 Some(x) => x,
546 None => {
547 // The leg/hop has disappeared. This is fine since the stream may have ended
548 // and been cleaned up while this `CtrlMsg::SendSendme` message was queued.
549 // It is possible that is a bug and this is an incorrect leg/hop number, but
550 // it's not currently possible to differentiate between an incorrect leg/hop
551 // number and a circuit hop that has been closed.
552 debug!(
553 "Could not send a stream-level SENDME on a hop that does not exist. Ignoring."
554 );
555 return Ok(None);
556 }
557 };
558
559 if !sendme_required {
560 // Nothing to do, so discard the SENDME.
561 return Ok(None);
562 }
563
564 let sendme = Sendme::new_empty();
565 let cell = AnyRelayMsgOuter::new(Some(stream_id), sendme.into());
566
567 let cell = SendRelayCell {
568 hop: Some(hop_num),
569 early: false,
570 cell,
571 };
572
573 Ok(Some(RunOnceCmdInner::Send {
574 leg: leg_id,
575 cell,
576 done: None,
577 }))
578 }
579 FlowCtrlMsg::Xon(rate) => Ok(Some(RunOnceCmdInner::MaybeSendXon {
580 rate,
581 hop,
582 stream_id,
583 })),
584 }
585 }
586 // TODO(conflux): this should specify which leg to send the msg on
587 // (currently we send it down the primary leg).
588 //
589 // This will involve updating ClientCIrc::send_raw_msg() to take a
590 // leg id argument (which is a breaking change.
591 #[cfg(feature = "send-control-msg")]
592 CtrlMsg::SendMsg { hop, msg, sender } => {
593 let Some((leg_id, hop_num)) = self.reactor.target_hop_to_hopnum_id(hop) else {
594 // Don't care if receiver goes away
595 let _ = sender.send(Err(bad_api_usage!("Unknown {hop:?}").into()));
596 return Ok(None);
597 };
598
599 let cell = AnyRelayMsgOuter::new(None, msg);
600 let cell = SendRelayCell {
601 hop: Some(hop_num),
602 early: false,
603 cell,
604 };
605
606 Ok(Some(RunOnceCmdInner::Send {
607 leg: leg_id,
608 cell,
609 done: Some(sender),
610 }))
611 }
612 // TODO(conflux): this should specify which leg to send the msg on
613 // (currently we send it down the primary leg)
614 #[cfg(feature = "send-control-msg")]
615 CtrlMsg::SendMsgAndInstallHandler {
616 msg,
617 handler,
618 sender,
619 } => Ok(Some(RunOnceCmdInner::SendMsgAndInstallHandler {
620 msg,
621 handler,
622 done: sender,
623 })),
624 CtrlMsg::FirstHopClockSkew { answer } => {
625 Ok(Some(RunOnceCmdInner::FirstHopClockSkew { answer }))
626 }
627 #[cfg(feature = "conflux")]
628 CtrlMsg::LinkCircuits { circuits, answer } => {
629 Ok(Some(RunOnceCmdInner::Link { circuits, answer }))
630 }
631 }
632 }
633
634 /// Handle a control command.
635 #[allow(clippy::needless_pass_by_value)] // Needed when conflux is enabled
636 pub(super) fn handle_cmd(&mut self, msg: CtrlCmd) -> StdResult<(), ReactorError> {
637 trace!(
638 tunnel_id = %self.reactor.tunnel_id,
639 msg = ?msg,
640 "reactor received control command"
641 );
642
643 match msg {
644 CtrlCmd::Shutdown => self.reactor.handle_shutdown().map(|_| ()),
645 #[cfg(feature = "hs-common")]
646 #[allow(unreachable_code)]
647 CtrlCmd::ExtendVirtual {
648 cell_crypto,
649 settings,
650 done,
651 } => {
652 let (outbound, inbound, binding) = cell_crypto;
653
654 // TODO HS: Perhaps this should describe the onion service, or
655 // describe why the virtual hop was added, or something?
656 let peer_id = path::HopDetail::Virtual;
657
658 let Ok(leg) = self.reactor.circuits.single_leg_mut() else {
659 // Don't care if the receiver goes away
660 let _ = done.send(Err(tor_error::bad_api_usage!(
661 "cannot extend multipath tunnel"
662 )
663 .into()));
664
665 return Ok(());
666 };
667
668 leg.add_hop(peer_id, outbound, inbound, binding, &settings)?;
669 let _ = done.send(Ok(()));
670
671 Ok(())
672 }
673 CtrlCmd::ResolveTargetHop { hop, done } => {
674 let _ = done.send(
675 self.reactor
676 .resolve_target_hop(hop)
677 .map_err(|_| crate::util::err::Error::NoSuchHop),
678 );
679 Ok(())
680 }
681 #[cfg(feature = "hs-service")]
682 CtrlCmd::AwaitStreamRequest {
683 cmd_checker,
684 incoming_sender,
685 hop,
686 done,
687 filter,
688 } => {
689 let Some((_, hop_num)) = self.reactor.target_hop_to_hopnum_id(hop) else {
690 let _ = done.send(Err(crate::Error::NoSuchHop));
691 return Ok(());
692 };
693 // TODO: At some point we might want to add a CtrlCmd for
694 // de-registering the handler. See comments on `allow_stream_requests`.
695 let handler = IncomingStreamRequestHandler {
696 incoming_sender,
697 cmd_checker,
698 hop_num: Some(hop_num),
699 filter,
700 };
701
702 let ret = self
703 .reactor
704 .cell_handlers
705 .set_incoming_stream_req_handler(handler);
706 let _ = done.send(ret); // don't care if the corresponding receiver goes away.
707
708 Ok(())
709 }
710 #[cfg(feature = "hs-service")]
711 CtrlCmd::GetBindingKey { hop, done } => {
712 let Some((leg_id, hop_num)) = self.reactor.target_hop_to_hopnum_id(hop) else {
713 let _ = done.send(Err(tor_error::internal!(
714 "Unknown TargetHop when getting binding key"
715 )
716 .into()));
717 return Ok(());
718 };
719 let Some(circuit) = self.reactor.circuits.leg(leg_id) else {
720 let _ = done.send(Err(tor_error::bad_api_usage!(
721 "Unknown circuit id {leg_id} when getting binding key"
722 )
723 .into()));
724 return Ok(());
725 };
726 // Get the binding key from the mutable state and send it back.
727 let key = circuit.mutable().binding_key(hop_num);
728 let _ = done.send(Ok(key));
729
730 Ok(())
731 }
732 #[cfg(test)]
733 CtrlCmd::AddFakeHop {
734 relay_cell_format,
735 fwd_lasthop,
736 rev_lasthop,
737 peer_id,
738 params,
739 done,
740 } => {
741 let Ok(leg) = self.reactor.circuits.single_leg_mut() else {
742 // Don't care if the receiver goes away
743 let _ = done.send(Err(tor_error::bad_api_usage!(
744 "cannot add fake hop to multipath tunnel"
745 )
746 .into()));
747
748 return Ok(());
749 };
750
751 leg.handle_add_fake_hop(
752 relay_cell_format,
753 fwd_lasthop,
754 rev_lasthop,
755 peer_id,
756 ¶ms,
757 done,
758 );
759
760 Ok(())
761 }
762 #[cfg(test)]
763 CtrlCmd::QuerySendWindow { hop, leg, done } => {
764 // Immediately invoked function means that errors will be sent to the channel.
765 let _ = done.send((|| {
766 let leg = self.reactor.circuits.leg_mut(leg).ok_or_else(|| {
767 bad_api_usage!("cannot query send window of non-existent circuit")
768 })?;
769
770 let hop = leg.hop_mut(hop).ok_or(bad_api_usage!(
771 "received QuerySendWindow for unknown hop {}",
772 hop.display()
773 ))?;
774
775 Ok(hop.send_window_and_expected_tags())
776 })());
777
778 Ok(())
779 }
780 #[cfg(feature = "conflux")]
781 CtrlCmd::ShutdownAndReturnCircuit { answer } => {
782 self.reactor.handle_shutdown_and_return_circuit(answer)
783 }
784 #[cfg(feature = "circ-padding-manual")]
785 CtrlCmd::SetPadder {
786 hop,
787 padder,
788 sender,
789 } => {
790 let result = self.reactor.set_padding_at_hop(hop, padder);
791 let _ = sender.send(result);
792 Ok(())
793 }
794 CtrlCmd::GetTunnelActivity { sender } => {
795 let count = self.reactor.circuits.tunnel_activity();
796 let _ = sender.send(count);
797 Ok(())
798 }
799 }
800 }
801}