1
#![cfg_attr(docsrs, feature(doc_cfg))]
2
#![doc = include_str!("../README.md")]
3
// @@ begin lint list maintained by maint/add_warning @@
4
#![allow(renamed_and_removed_lints)] // @@REMOVE_WHEN(ci_arti_stable)
5
#![allow(unknown_lints)] // @@REMOVE_WHEN(ci_arti_nightly)
6
#![warn(missing_docs)]
7
#![warn(noop_method_call)]
8
#![warn(unreachable_pub)]
9
#![warn(clippy::all)]
10
#![deny(clippy::await_holding_lock)]
11
#![deny(clippy::cargo_common_metadata)]
12
#![deny(clippy::cast_lossless)]
13
#![deny(clippy::checked_conversions)]
14
#![allow(clippy::cognitive_complexity)] // See arti#2556
15
#![deny(clippy::debug_assert_with_mut_call)]
16
#![deny(clippy::exhaustive_enums)]
17
#![deny(clippy::exhaustive_structs)]
18
#![deny(clippy::expl_impl_clone_on_copy)]
19
#![deny(clippy::fallible_impl_from)]
20
#![deny(clippy::implicit_clone)]
21
#![deny(clippy::large_stack_arrays)]
22
#![warn(clippy::manual_ok_or)]
23
#![deny(clippy::missing_docs_in_private_items)]
24
#![warn(clippy::needless_borrow)]
25
#![warn(clippy::needless_pass_by_value)]
26
#![warn(clippy::option_option)]
27
#![warn(clippy::or_fun_call)]
28
#![deny(clippy::print_stderr)]
29
#![deny(clippy::print_stdout)]
30
#![warn(clippy::rc_buffer)]
31
#![deny(clippy::ref_option_ref)]
32
#![warn(clippy::semicolon_if_nothing_returned)]
33
#![warn(clippy::trait_duplication_in_bounds)]
34
#![deny(clippy::unchecked_time_subtraction)]
35
#![deny(clippy::unnecessary_wraps)]
36
#![warn(clippy::unseparated_literal_suffix)]
37
#![deny(clippy::unwrap_used)]
38
#![deny(clippy::mod_module_files)]
39
#![allow(clippy::let_unit_value)] // This can reasonably be done for explicitness
40
#![allow(clippy::uninlined_format_args)]
41
#![allow(clippy::significant_drop_in_scrutinee)] // arti/-/merge_requests/588/#note_2812945
42
#![allow(clippy::result_large_err)] // temporary workaround for arti#587
43
#![allow(clippy::needless_raw_string_hashes)] // complained-about code is fine, often best
44
#![allow(clippy::needless_lifetimes)] // See arti#1765
45
#![allow(mismatched_lifetime_syntaxes)] // temporary workaround for arti#2060
46
#![allow(clippy::collapsible_if)] // See arti#2342
47
#![deny(clippy::unused_async)]
48
#![deny(clippy::string_slice)] // See arti#2571
49
#![allow(recursion_depth_exceeding_limit)] // arti#2715, rust/issues/159228
50
//! <!-- @@ end lint list maintained by maint/add_warning @@ -->
51

            
52
pub mod builder;
53
mod config;
54
mod err;
55
mod event;
56
pub mod factory;
57
mod mgr;
58
#[cfg(test)]
59
mod testing;
60
pub mod transport;
61

            
62
use futures::StreamExt;
63
use futures::select_biased;
64
use std::result::Result as StdResult;
65
use std::sync::{Arc, Weak};
66
use std::time::Duration;
67
use tor_config::ReconfigureError;
68
use tor_error::error_report;
69
use tor_linkspec::{ChanTarget, OwnedChanTarget};
70
use tor_netdir::{NetDirProvider, params::NetParameters};
71
use tor_proto::channel::Channel;
72
#[cfg(feature = "experimental-api")]
73
use tor_proto::memquota::ChannelAccount;
74
use tor_proto::memquota::ToplevelAccount;
75
use tor_rtcompat::SpawnExt;
76
use tracing::debug;
77
use tracing::instrument;
78
use void::{ResultVoidErrExt, Void};
79

            
80
#[cfg(feature = "relay")]
81
use {
82
    async_trait::async_trait, safelog::Sensitive, tor_proto::relay::CreateRequestHandler,
83
    tor_proto::relay::channel_provider::ChannelProvider,
84
};
85

            
86
pub use err::Error;
87

            
88
pub use config::{ChannelConfig, ChannelConfigBuilder, ProxyProtocol};
89
pub use mgr::ChanMgrConfig;
90

            
91
use tor_rtcompat::Runtime;
92

            
93
/// A Result as returned by this crate.
94
pub type Result<T> = std::result::Result<T, Error>;
95

            
96
use crate::factory::BootstrapReporter;
97
pub use event::{ConnBlockage, ConnStatus, ConnStatusEvents};
98
use tor_rtcompat::scheduler::{TaskHandle, TaskSchedule};
99

            
100
/// An object that remembers a set of live channels, and launches new ones on
101
/// request.
102
///
103
/// Use the [`ChanMgr::get_or_launch`] function to create a new [`Channel`], or
104
/// get one if it exists.  (For a slightly lower-level API that does no caching,
105
/// see [`ChannelFactory`](factory::ChannelFactory) and its implementors.
106
///
107
/// Each channel is kept open as long as there is a reference to it, or
108
/// something else (such as the relay or a network error) kills the channel.
109
///
110
/// After a `ChanMgr` launches a channel, it keeps a reference to it until that
111
/// channel has been unused (that is, had no circuits attached to it) for a
112
/// certain amount of time. (Currently this interval is chosen randomly from
113
/// between 180-270 seconds, but this is an implementation detail that may change
114
/// in the future.)
115
pub struct ChanMgr<R: Runtime> {
116
    /// Internal channel manager object that does the actual work.
117
    ///
118
    /// ## How this is built
119
    ///
120
    /// This internal manager is parameterized over an
121
    /// [`mgr::AbstractChannelFactory`], which here is instantiated with a [`factory::CompoundFactory`].
122
    /// The `CompoundFactory` itself holds:
123
    ///   * A `dyn` [`factory::AbstractPtMgr`] that can provide a `dyn`
124
    ///     [`factory::ChannelFactory`] for each supported pluggable transport.
125
    ///     This starts out as `None`, but can be replaced with [`ChanMgr::set_pt_mgr`].
126
    ///     The `TorClient` code currently sets this using `tor_ptmgr::PtMgr`.
127
    ///     `PtMgr` currently returns `ChannelFactory` implementations that are
128
    ///     built using [`transport::proxied::ExternalProxyPlugin`], which implements
129
    ///     [`transport::TransportImplHelper`], which in turn is wrapped into a
130
    ///     `ChanBuilder` to implement `ChannelFactory`.
131
    ///   * A generic [`factory::ChannelFactory`] that it uses for everything else
132
    ///     We instantiate this with a
133
    ///     [`builder::ChanBuilder`] using a [`transport::default::DefaultTransport`].
134
    // This type is a bit long, but I think it's better to just state it here explicitly rather than
135
    // hiding parts of it behind a type alias to make it look nicer.
136
    mgr: mgr::AbstractChanMgr<
137
        factory::CompoundFactory<builder::ChanBuilder<R, transport::DefaultTransport<R>>>,
138
    >,
139

            
140
    /// Stream of [`ConnStatus`] events.
141
    bootstrap_status: event::ConnStatusEvents,
142

            
143
    /// The runtime. Needed to possibly spawn tasks.
144
    #[allow(unused)] // Relay use this, not client yet. Keep it here instead of gating.
145
    runtime: R,
146
}
147

            
148
/// Description of how we got a channel.
149
#[non_exhaustive]
150
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
151
pub enum ChanProvenance {
152
    /// This channel was newly launched, or was in progress and finished while
153
    /// we were waiting.
154
    NewlyCreated,
155
    /// This channel already existed when we asked for it.
156
    Preexisting,
157
}
158

            
159
/// Dormancy state, as far as the channel manager is concerned
160
///
161
/// This is usually derived in higher layers from `arti_client::DormantMode`.
162
#[non_exhaustive]
163
#[derive(Debug, Default, Copy, Clone, Eq, PartialEq)]
164
pub enum Dormancy {
165
    /// Not dormant
166
    ///
167
    /// Channels will operate normally.
168
    #[default]
169
    Active,
170
    /// Totally dormant
171
    ///
172
    /// Channels will not perform any spontaneous activity (eg, netflow padding)
173
    Dormant,
174
}
175

            
176
/// The usage that we have in mind when requesting a channel.
177
///
178
/// A channel may be used in multiple ways.  Each time a channel is requested
179
/// from `ChanMgr` a separate `ChannelUsage` is passed in to tell the `ChanMgr`
180
/// how the channel will be used this time.
181
///
182
/// To be clear, the `ChannelUsage` is aspect of a _request_ for a channel, and
183
/// is not an immutable property of the channel itself.
184
///
185
/// This type is obtained from a `tor_circmgr::usage::SupportedCircUsage` in
186
/// `tor_circmgr::usage`, and it has roughly the same set of variants.
187
#[derive(Clone, Debug, Copy, Eq, PartialEq)]
188
#[non_exhaustive]
189
pub enum ChannelUsage {
190
    /// Requesting a channel to use for BEGINDIR-based non-anonymous directory
191
    /// connections.
192
    Dir,
193

            
194
    /// Requesting a channel to transmit user traffic (including exit traffic)
195
    /// over the network.
196
    ///
197
    /// This includes the case where we are constructing a circuit preemptively,
198
    /// and _planning_ to use it for user traffic later on.
199
    UserTraffic,
200

            
201
    /// Requesting a channel that the caller does not plan to used at all, or
202
    /// which it plans to use only for testing circuits.
203
    UselessCircuit,
204
}
205

            
206
impl<R: Runtime> ChanMgr<R> {
207
    /// Construct a new channel manager.
208
    ///
209
    /// A new `ChannelAccount` will be made from `memquota`, for each Channel.
210
    ///
211
    /// The `ChannelAccount` is used for data associated with this channel.
212
    ///
213
    /// This does *not* (currently) include downstream outbound data
214
    /// (ie, data processed by the channel implementation here,
215
    /// awaiting TLS processing and actual transmission).
216
    /// In any case we try to keep those buffers small.
217
    ///
218
    /// The ChannelAccount *does* track upstream outbound data
219
    /// (ie, data processed by a circuit, but not yet by the channel),
220
    /// even though that data relates to a specific circuit.
221
    /// TODO #1652 use `CircuitAccount` for circuit->channel queue.
222
    ///
223
    /// # Usage note
224
    ///
225
    /// For the manager to work properly, you will need to call `ChanMgr::launch_background_tasks`.
226
    ///
227
    /// The `keymgr` is only needed for a relay which is used for authenticating its channel to
228
    /// other relays. Pass `None` for a client.
229
18
    pub fn new(
230
18
        runtime: R,
231
18
        config: ChanMgrConfig,
232
18
        dormancy: Dormancy,
233
18
        netparams: &NetParameters,
234
18
        memquota: ToplevelAccount,
235
18
    ) -> Result<Self>
236
18
    where
237
18
        R: 'static,
238
    {
239
18
        let (sender, receiver) = event::channel();
240
18
        let sender = Arc::new(std::sync::Mutex::new(sender));
241
18
        let reporter = BootstrapReporter(sender);
242
18
        let transport =
243
18
            transport::DefaultTransport::new(runtime.clone(), config.cfg.outbound_proxy.clone());
244
        cfg_if::cfg_if! {
245
            if #[cfg(feature = "relay")] {
246
18
                let builder = if let Some(auth_material) = &config.auth_material {
247
                    builder::ChanBuilder::new_relay(runtime.clone(), transport, auth_material.clone(), config.my_addrs, None)?
248
                } else {
249
                    // Yes, clients can have the "relay" feature enabled (unit tests).
250
18
                    builder::ChanBuilder::new_client(runtime.clone(), transport)
251
                };
252
            } else {
253
                let builder =  builder::ChanBuilder::new_client(runtime.clone(), transport);
254
            }
255
        };
256

            
257
18
        let factory = factory::CompoundFactory::new(
258
18
            Arc::new(builder),
259
            #[cfg(feature = "pt-client")]
260
18
            None,
261
        );
262

            
263
        // Warn if outbound_proxy is configured to a non-loopback address
264
18
        if let Some(ref proxy) = config.cfg.outbound_proxy {
265
            if !proxy.is_loopback() {
266
                tracing::warn!(
267
                    proxy_addr = %proxy,
268
                    "outbound_proxy is configured to a non-loopback address; \
269
                     this may expose Tor traffic to an untrusted intermediate"
270
                );
271
            }
272
18
        }
273

            
274
18
        let mgr =
275
18
            mgr::AbstractChanMgr::new(factory, config.cfg, dormancy, netparams, reporter, memquota);
276

            
277
18
        Ok(ChanMgr {
278
18
            mgr,
279
18
            bootstrap_status: receiver,
280
18
            runtime,
281
18
        })
282
18
    }
283

            
284
    /// Launch the periodic daemon tasks required by the manager to function properly.
285
    ///
286
    /// Returns a [`TaskHandle`] that can be used to manage
287
    /// those daemon tasks that poll periodically.
288
    #[instrument(level = "trace", skip_all)]
289
    pub fn launch_background_tasks(
290
        self: &Arc<Self>,
291
        runtime: &R,
292
        netdir: Arc<dyn NetDirProvider>,
293
    ) -> Result<Vec<TaskHandle>> {
294
        runtime
295
            .spawn(Self::continually_update_channels_config(
296
                Arc::downgrade(self),
297
                netdir,
298
            ))
299
            .map_err(|e| Error::from_spawn("channels config task", e))?;
300

            
301
        let (sched, handle) = TaskSchedule::new(runtime.clone());
302
        runtime
303
            .spawn(Self::continually_expire_channels(
304
                sched,
305
                Arc::downgrade(self),
306
            ))
307
            .map_err(|e| Error::from_spawn("channel expiration task", e))?;
308
        Ok(vec![handle])
309
    }
310

            
311
    /// Build a channel for an incoming stream.
312
    ///
313
    /// The `my_addrs` are the IP address(es) that are advertised by the relay in the consensus. We
314
    /// need to pass them so they can be sent in the NETINFO cell.
315
    ///
316
    /// The channel may or may not be authenticated. This method will wait until the channel is
317
    /// usable, and may return an error if we already have an existing channel to this peer.
318
    #[cfg(feature = "relay")]
319
    pub async fn handle_incoming(
320
        &self,
321
        src: Sensitive<std::net::SocketAddr>,
322
        stream: <R as tor_rtcompat::NetStreamProvider>::Stream,
323
    ) -> Result<Arc<Channel>> {
324
        let result = self.mgr.handle_incoming(src, stream).await;
325

            
326
        #[cfg(feature = "metrics")]
327
        self.mgr.metrics.increment_inbound_channels_built(&result);
328

            
329
        result
330
    }
331

            
332
    /// Try to get a suitable channel to the provided `target`,
333
    /// launching one if one does not exist.
334
    ///
335
    /// This function does not guarantee that the returned channel
336
    /// satisfies all of the properties of `target`. For example if an
337
    /// existing channel is returned, it might not be connected to any
338
    /// of the addresses specified in `target`.
339
    // ^ see https://gitlab.torproject.org/tpo/core/arti/-/issues/2344
340
    ///
341
    /// If there is already a channel launch attempt in progress, this
342
    /// function will wait until that launch is complete, and succeed
343
    /// or fail depending on its outcome.
344
    #[instrument(level = "trace", skip_all)]
345
    pub async fn get_or_launch<T: ChanTarget + ?Sized>(
346
        &self,
347
        target: &T,
348
        usage: ChannelUsage,
349
    ) -> Result<(Arc<Channel>, ChanProvenance)> {
350
        let targetinfo = OwnedChanTarget::from_chan_target(target);
351

            
352
        let (chan, provenance) = self.mgr.get_or_launch(targetinfo, usage).await?;
353
        // Double-check the match to make sure that the RSA identity is
354
        // what we wanted too.
355
        chan.check_match(target)
356
            .map_err(|e| Error::from_proto_no_skew(e, target))?;
357
        Ok((chan, provenance))
358
    }
359

            
360
    /// Return a stream of [`ConnStatus`] events to tell us about changes
361
    /// in our ability to connect to the internet.
362
    ///
363
    /// Note that this stream can be lossy: the caller will not necessarily
364
    /// observe every event on the stream
365
    pub fn bootstrap_events(&self) -> ConnStatusEvents {
366
        self.bootstrap_status.clone()
367
    }
368

            
369
    /// Expire all channels that have been unused for too long.
370
    ///
371
    /// Return the duration from now until next channel expires.
372
    pub fn expire_channels(&self) -> Duration {
373
        self.mgr.expire_channels()
374
    }
375

            
376
    /// Notifies the chanmgr to be dormant like dormancy
377
    pub fn set_dormancy(
378
        &self,
379
        dormancy: Dormancy,
380
        netparams: Arc<dyn AsRef<NetParameters>>,
381
    ) -> StdResult<(), tor_error::Bug> {
382
        self.mgr.set_dormancy(dormancy, netparams)
383
    }
384

            
385
    /// Reconfigure all channels
386
    pub fn reconfigure(
387
        &self,
388
        config: &ChannelConfig,
389
        how: tor_config::Reconfigure,
390
        netparams: Arc<dyn AsRef<NetParameters>>,
391
    ) -> StdResult<(), ReconfigureError> {
392
        if how == tor_config::Reconfigure::CheckAllOrNothing {
393
            // Since `self.mgr.reconfigure` returns an error type of `Bug` and not
394
            // `ReconfigureError` (see check below), the reconfigure should only fail due to bugs.
395
            // This means we can return `Ok` here since there should never be an error with the
396
            // provided `config` values.
397
            return Ok(());
398
        }
399

            
400
        let r = self.mgr.reconfigure(config, netparams);
401

            
402
        // Check that `self.mgr.reconfigure` returns an error type of `Bug` (see comment above).
403
        let _: Option<&tor_error::Bug> = r.as_ref().err();
404

            
405
        Ok(r?)
406
    }
407

            
408
    /// Replace the transport registry with one that may know about
409
    /// more transports.
410
    ///
411
    /// Note that the [`ChannelFactory`](factory::ChannelFactory) instances returned by `ptmgr` are
412
    /// required to time-out channels that take too long to build.  You'll get
413
    /// this behavior by default if the factories implement [`ChannelFactory`](factory::ChannelFactory) using
414
    /// [`transport::proxied::ExternalProxyPlugin`], which `tor-ptmgr` does.
415
    #[cfg(feature = "pt-client")]
416
    pub fn set_pt_mgr(&self, ptmgr: Arc<dyn factory::AbstractPtMgr + 'static>) {
417
        self.mgr.with_mut_builder(|f| f.replace_ptmgr(ptmgr));
418
    }
419

            
420
    /// Replace the relay auth material used for building new channels.
421
    ///
422
    /// This rebuilds the internal channel builder with the provided `auth_material`, which includes a
423
    /// new TLS cert and key. Existing channels are not affected, only newly created channels will
424
    /// use the new keys.
425
    #[cfg(feature = "relay")]
426
    pub fn set_relay_auth_material(
427
        &self,
428
        auth_material: Arc<tor_proto::RelayChannelAuthMaterial>,
429
    ) -> Result<()> {
430
        let mut result = Ok(());
431
        self.mgr.with_mut_builder(|f| {
432
            match f
433
                .default_factory()
434
                .rebuild_with_auth_material(auth_material)
435
            {
436
                Ok(b) => f.replace_default_factory(Arc::new(b)),
437
                Err(e) => result = Err(e),
438
            }
439
        });
440
        result
441
    }
442

            
443
    /// This will be used to handle CREATE* requests on channels.
444
    ///
445
    /// This handler will only be used for new channels, not existing channels.
446
    ///
447
    /// This will *not* be updated in any way by the channel manager,
448
    /// for example by a netdir update or when any keys change.
449
    /// The caller must handle this.
450
    /// The idea is that the channel manager shouldn't need to deal with circuit-specific stuff.
451
    ///
452
    /// It's expected to only ever call this once.
453
    /// Ideally it would be an `Option` in the constructor,
454
    /// but we don't want conditionally-compiled constructor arguments,
455
    /// and the [`CreateRequestHandler`] requires a [`dyn ChannelProvider`]
456
    /// which is typically this [`ChanMgr`] itself.
457
    #[cfg(feature = "relay")]
458
    pub fn set_create_request_handler(&self, handler: Arc<CreateRequestHandler>) -> Result<()> {
459
        let mut result = Ok(());
460
        self.mgr.with_mut_builder(|f| {
461
            match f
462
                .default_factory()
463
                .rebuild_with_create_request_handler(handler)
464
            {
465
                Ok(b) => f.replace_default_factory(Arc::new(b)),
466
                Err(e) => result = Err(e),
467
            }
468
        });
469
        result
470
    }
471

            
472
    /// Try to create a new, unmanaged channel to `target`.
473
    ///
474
    /// Unlike [`get_or_launch`](ChanMgr::get_or_launch), this function always
475
    /// creates a new channel, never retries transient failure, and does not
476
    /// register this channel with the `ChanMgr`.
477
    ///
478
    /// Generally you should not use this function; `get_or_launch` is usually a
479
    /// better choice.  This function is the right choice if, for whatever
480
    /// reason, you need to manage the lifetime of the channel you create, and
481
    /// make sure that no other code with access to this `ChanMgr` will be able
482
    /// to use the channel.
483
    #[cfg(feature = "experimental-api")]
484
    #[instrument(level = "trace", skip_all)]
485
    pub async fn build_unmanaged_channel(
486
        &self,
487
        target: impl tor_linkspec::IntoOwnedChanTarget,
488
        memquota: ChannelAccount,
489
    ) -> Result<Arc<Channel>> {
490
        use factory::ChannelFactory as _;
491
        let target = target.to_owned();
492

            
493
        self.mgr
494
            .channels
495
            .builder()
496
            .connect_via_transport(&target, self.mgr.reporter.clone(), memquota)
497
            .await
498
    }
499

            
500
    /// Watch for things that ought to change the configuration of all channels in the client
501
    ///
502
    /// Currently this handles enabling and disabling channel padding.
503
    ///
504
    /// This is a daemon task that runs indefinitely in the background,
505
    /// and exits when we find that `chanmgr` is dropped.
506
    #[instrument(level = "trace", skip_all)]
507
    async fn continually_update_channels_config(
508
        self_: Weak<Self>,
509
        netdir: Arc<dyn NetDirProvider>,
510
    ) {
511
        use tor_netdir::DirEvent as DE;
512
        let mut netdir_stream = netdir.events().fuse();
513
        let netdir = {
514
            let weak = Arc::downgrade(&netdir);
515
            drop(netdir);
516
            weak
517
        };
518
        let termination_reason: std::result::Result<Void, &str> = async move {
519
            loop {
520
                select_biased! {
521
                    direvent = netdir_stream.next() => {
522
                        let direvent = direvent.ok_or("EOF on netdir provider event stream")?;
523
                        if ! matches!(direvent, DE::NewConsensus) { continue };
524
                        let self_ = self_.upgrade().ok_or("channel manager gone away")?;
525
                        let netdir = netdir.upgrade().ok_or("netdir gone away")?;
526
                        let netparams = netdir.params();
527
                        self_.mgr.update_netparams(netparams).map_err(|e| {
528
                            error_report!(e, "continually_update_channels_config: failed to process!");
529
                            "error processing netdir"
530
                        })?;
531
                    }
532
                }
533
            }
534
        }
535
        .await;
536
        debug!(
537
            "continually_update_channels_config: shutting down: {}",
538
            termination_reason.void_unwrap_err()
539
        );
540
    }
541

            
542
    /// Periodically expire any channels that have been unused beyond
543
    /// the maximum duration allowed.
544
    ///
545
    /// Exist when we find that `chanmgr` is dropped
546
    ///
547
    /// This is a daemon task that runs indefinitely in the background
548
    #[instrument(level = "trace", skip_all)]
549
    async fn continually_expire_channels(mut sched: TaskSchedule<R>, chanmgr: Weak<Self>) {
550
        while sched.next().await.is_some() {
551
            let Some(cm) = Weak::upgrade(&chanmgr) else {
552
                // channel manager is closed.
553
                return;
554
            };
555
            let delay = cm.expire_channels();
556
            // This will sometimes be an underestimate, but it's no big deal; we just sleep some more.
557
            sched.fire_in(delay);
558
        }
559
    }
560
}
561

            
562
#[cfg(feature = "relay")]
563
#[async_trait]
564
impl<R: Runtime> ChannelProvider for ChanMgr<R> {
565
    type BuildSpec = OwnedChanTarget;
566

            
567
    fn get_or_launch(
568
        self: Arc<Self>,
569
        reactor_id: tor_proto::circuit::UniqId,
570
        target: Self::BuildSpec,
571
        tx: tor_proto::relay::channel_provider::OutboundChanSender,
572
    ) -> tor_proto::Result<()> {
573
        use tor_error::into_internal;
574

            
575
        debug!("Get or launch channel to {target} for circuit reactor {reactor_id}");
576

            
577
        let chanmgr = self.clone();
578
        self.runtime
579
            .spawn(async move {
580
                let r = chanmgr
581
                    .mgr
582
                    .get_or_launch(target, ChannelUsage::UserTraffic)
583
                    .await
584
                    .map_err(|e| tor_proto::Error::ChanProto(e.to_string())); // Is it a ChanProto?
585
                // Send back the channel.
586
                tx.send(r.map(|(chan, _)| chan));
587
            })
588
            .map_err(into_internal!("Failed to launch channel provider task"))?;
589

            
590
        Ok(())
591
    }
592
}