1
//! Code for talking directly (over a TLS connection) to a Tor client or relay.
2
//!
3
//! Channels form the basis of the rest of the Tor protocol: they are
4
//! the only way for two Tor instances to talk.
5
//!
6
//! Channels are not useful directly for application requests: after
7
//! making a channel, it needs to get used to build circuits, and the
8
//! circuits are used to anonymize streams.  The streams are the
9
//! objects corresponding to directory requests.
10
//!
11
//! In general, you shouldn't try to manage channels on your own;
12
//! use the `tor-chanmgr` crate instead.
13
//!
14
//! To launch a channel:
15
//!
16
//!  * Create a TLS connection as an object that implements AsyncRead +
17
//!    AsyncWrite + StreamOps, and pass it to a channel builder. This will
18
//!    yield an [crate::client::channel::handshake::ClientInitiatorHandshake] that represents
19
//!    the state of the handshake.
20
//!  * Call [crate::client::channel::handshake::ClientInitiatorHandshake::connect] on the result
21
//!    to negotiate the rest of the handshake.  This will verify
22
//!    syntactic correctness of the handshake, but not its cryptographic
23
//!    integrity.
24
//!  * Call handshake::UnverifiedChannel::check on the result.  This
25
//!    finishes the cryptographic checks.
26
//!  * Call handshake::VerifiedChannel::finish on the result. This
27
//!    completes the handshake and produces an open channel and Reactor.
28
//!  * Launch an asynchronous task to call the reactor's run() method.
29
//!
30
//! One you have a running channel, you can create circuits on it with
31
//! its [Channel::new_tunnel] method.  See
32
//! [crate::client::circuit::PendingClientTunnel] for information on how to
33
//! proceed from there.
34
//!
35
//! # Design
36
//!
37
//! For now, this code splits the channel into two pieces: a "Channel"
38
//! object that can be used by circuits to write cells onto the
39
//! channel, and a "Reactor" object that runs as a task in the
40
//! background, to read channel cells and pass them to circuits as
41
//! appropriate.
42
//!
43
//! I'm not at all sure that's the best way to do that, but it's what
44
//! I could think of.
45
//!
46
//! # Limitations
47
//!
48
//! TODO: There is no rate limiting or fairness.
49

            
50
/// The size of the channel buffer for communication between `Channel` and its reactor.
51
pub const CHANNEL_BUFFER_SIZE: usize = 128;
52

            
53
pub(crate) mod circmap;
54
pub(crate) mod handler;
55
pub(crate) mod handshake;
56
pub mod kist;
57
mod msg;
58
pub mod padding;
59
pub mod params;
60
mod reactor;
61
mod unique_id;
62

            
63
pub use crate::channel::params::*;
64
pub(crate) use crate::channel::reactor::Reactor;
65
use crate::channel::reactor::{BoxedChannelSink, BoxedChannelStream};
66
pub use crate::channel::unique_id::UniqId;
67
use crate::client::circuit::PendingClientTunnel;
68
use crate::client::circuit::padding::{PaddingController, QueuedCellPaddingInfo};
69
use crate::memquota::{ChannelAccount, CircuitAccount, SpecificAccount as _};
70
use crate::peer::PeerInfo;
71
use crate::util::err::ChannelClosed;
72
use crate::util::oneshot_broadcast;
73
use crate::util::timeout::TimeoutEstimator;
74
use crate::util::ts::AtomicOptTimestamp;
75
use crate::{ClockSkew, client};
76
use crate::{Error, Result};
77
use cfg_if::cfg_if;
78
use reactor::BoxedChannelStreamOps;
79
use safelog::{MaybeSensitive, sensitive as sv};
80
use std::future::{Future, IntoFuture};
81
use std::net::IpAddr;
82
use std::pin::Pin;
83
use std::sync::{Mutex, MutexGuard};
84
use std::time::Duration;
85
use tor_cell::chancell::ChanMsg;
86
use tor_cell::chancell::{AnyChanCell, CircId, msg::Netinfo, msg::PaddingNegotiate};
87
use tor_error::internal;
88
use tor_linkspec::{HasRelayIds, OwnedChanTarget};
89
use tor_memquota::mq_queue::{self, ChannelSpec as _, MpscSpec};
90
use tor_rtcompat::{CoarseTimeProvider, DynTimeProvider, Runtime, SleepProvider};
91

            
92
#[cfg(feature = "circ-padding")]
93
use tor_async_utils::counting_streams::{self, CountingSink, CountingStream};
94

            
95
#[cfg(feature = "relay")]
96
use {
97
    crate::channel::reactor::CreateRequestHandlerAndData, crate::circuit::CircuitRxReceiver,
98
    crate::relay::channel::create_handler::CreateRequestHandler,
99
    tor_llcrypto::pk::ed25519::Ed25519Identity, tor_llcrypto::pk::rsa::RsaIdentity,
100
};
101

            
102
/// Imports that are re-exported pub if feature `testing` is enabled
103
///
104
/// Putting them together in a little module like this allows us to select the
105
/// visibility for all of these things together.
106
mod testing_exports {
107
    #![allow(unreachable_pub)]
108
    pub use super::reactor::CtrlMsg;
109
    pub use crate::circuit::celltypes::CreateResponse;
110
}
111
#[cfg(feature = "testing")]
112
pub use testing_exports::*;
113
#[cfg(not(feature = "testing"))]
114
use testing_exports::*;
115

            
116
use asynchronous_codec;
117
use futures::channel::mpsc;
118
use futures::io::{AsyncRead, AsyncWrite};
119
use oneshot_fused_workaround as oneshot;
120

            
121
use educe::Educe;
122
use futures::{FutureExt as _, Sink};
123
use std::result::Result as StdResult;
124
use std::sync::Arc;
125
use std::task::{Context, Poll};
126

            
127
use tracing::{instrument, trace};
128

            
129
// reexport
130
pub use super::client::channel::handshake::ClientInitiatorHandshake;
131
#[cfg(feature = "relay")]
132
pub use super::relay::channel::handshake::RelayInitiatorHandshake;
133
pub(crate) use crate::channel::handler::{ClogDigest, SlogDigest};
134
use crate::channel::unique_id::CircUniqIdContext;
135

            
136
use kist::KistParams;
137

            
138
/// This indicate what type of channel it is. It allows us to decide for the correct channel cell
139
/// state machines and authentication process (if any).
140
///
141
/// It is created when a channel is requested for creation which means the subsystem wanting to
142
/// open a channel needs to know what type it wants.
143
#[derive(Clone, Copy, Debug, derive_more::Display)]
144
#[non_exhaustive]
145
pub enum ChannelType {
146
    /// Client: Initiated from a client to a relay. Client is unauthenticated and relay is
147
    /// authenticated.
148
    ClientInitiator,
149
    /// Relay: Initiating as a relay to a relay. Both sides are authenticated.
150
    RelayInitiator,
151
    /// Relay: Responding as a relay to a relay or client. Authenticated or Unauthenticated.
152
    RelayResponder {
153
        /// Indicate if the channel is authenticated. Responding as a relay can be either from a
154
        /// Relay (authenticated) or a Client/Bridge (Unauthenticated). We only know this
155
        /// information once the handshake is completed.
156
        ///
157
        /// This side is always authenticated, the other side can be if a relay or not if
158
        /// bridge/client. This is set to false unless we end up authenticating the other side
159
        /// meaning a relay.
160
        authenticated: bool,
161
    },
162
}
163

            
164
impl ChannelType {
165
    /// Set that this channel type is now authenticated. This only applies to RelayResponder.
166
    pub(crate) fn set_authenticated(&mut self) {
167
        if let Self::RelayResponder { authenticated } = self {
168
            *authenticated = true;
169
        }
170
    }
171
}
172

            
173
/// A channel cell frame used for sending and receiving cells on a channel. The handler takes care
174
/// of the cell codec transition depending in which state the channel is.
175
///
176
/// ChannelFrame is used to basically handle all in and outbound cells on a channel for its entire
177
/// lifetime.
178
pub(crate) type ChannelFrame<T> = asynchronous_codec::Framed<T, handler::ChannelCellHandler>;
179

            
180
/// An entry in a channel's queue of cells to be flushed.
181
pub(crate) type ChanCellQueueEntry = (AnyChanCell, Option<QueuedCellPaddingInfo>);
182

            
183
/// Helper: Return a new channel frame [ChannelFrame] from an object implementing AsyncRead + AsyncWrite. In the
184
/// tor context, it is always a TLS stream.
185
///
186
/// The ty (type) argument needs to be able to transform into a [handler::ChannelCellHandler] which would
187
/// generally be a [ChannelType].
188
94
pub(crate) fn new_frame<T, I>(tls: T, ty: I) -> ChannelFrame<T>
189
94
where
190
94
    T: AsyncRead + AsyncWrite,
191
94
    I: Into<handler::ChannelCellHandler>,
192
{
193
94
    let mut framed = asynchronous_codec::Framed::new(tls, ty.into());
194
94
    framed.set_send_high_water_mark(32 * 1024);
195
94
    framed
196
94
}
197

            
198
/// Canonical state between this channel and its peer. This is inferred from the [`Netinfo`]
199
/// received during the channel handshake.
200
///
201
/// A connection is "canonical" if the TCP connection's peer IP address matches an address
202
/// that the relay itself claims in its [`Netinfo`] cell.
203
#[derive(Debug)]
204
pub(crate) struct Canonicity {
205
    /// The peer has proven this connection is canonical for its address: at least one NETINFO "my
206
    /// address" matches the observed TCP peer address.
207
    pub(crate) peer_is_canonical: bool,
208
    /// We appear canonical from the peer's perspective: its NETINFO "other address" matches our
209
    /// advertised relay address.
210
    pub(crate) canonical_to_peer: bool,
211
}
212

            
213
impl Canonicity {
214
    /// Using a [`Netinfo`], build the canonicity object with the given addresses.
215
    ///
216
    /// The `my_addrs` are the advertised address of this relay or empty if a client/bridge as they
217
    /// do not advertise or expose a reachable address.
218
    ///
219
    /// The `peer_addr` is the IP address we believe the peer has. In other words, it is either the
220
    /// IP we used to connect to or the address we see in the accept() phase of the connection.
221
    ///
222
    /// It can be None if we used a non-IP address to connect to the peer (PT).
223
50
    pub(crate) fn from_netinfo(
224
50
        netinfo: &Netinfo,
225
50
        my_addrs: &[IpAddr],
226
50
        peer_addr: Option<IpAddr>,
227
50
    ) -> Self {
228
        Self {
229
            // The "other addr" (our address as seen by the peer) matches the one we advertised.
230
50
            canonical_to_peer: netinfo
231
50
                .their_addr()
232
52
                .is_some_and(|a: &IpAddr| my_addrs.contains(a)),
233
            // The "my addresses" (the peer addresses that it claims to have) matches the one we
234
            // see on the connection or that we attempted to connect to.
235
50
            peer_is_canonical: peer_addr
236
52
                .map(|a| netinfo.my_addrs().contains(&a))
237
50
                .unwrap_or_default(),
238
        }
239
50
    }
240

            
241
    /// Construct a fully canonical object.
242
    #[cfg(any(test, feature = "testing"))]
243
1664
    pub(crate) fn new_canonical() -> Self {
244
1664
        Self {
245
1664
            peer_is_canonical: true,
246
1664
            canonical_to_peer: true,
247
1664
        }
248
1664
    }
249
}
250

            
251
/// An open client channel, ready to send and receive Tor cells.
252
///
253
/// A channel is a direct connection to a Tor relay, implemented using TLS.
254
///
255
/// This struct is a frontend that can be used to send cells
256
/// and otherwise control the channel.  The main state is
257
/// in the Reactor object.
258
///
259
/// (Users need a mutable reference because of the types in `Sink`, and
260
/// ultimately because `cell_tx: mpsc::Sender` doesn't work without mut.
261
///
262
/// # Channel life cycle
263
///
264
/// Channels can be created directly here through a channel builder (client or relay) API.
265
/// For a higher-level API (with better support for TLS, pluggable transports,
266
/// and channel reuse) see the `tor-chanmgr` crate.
267
///
268
/// After a channel is created, it will persist until it is closed in one of
269
/// four ways:
270
///    1. A remote error occurs.
271
///    2. The other side of the channel closes the channel.
272
///    3. Someone calls [`Channel::terminate`] on the channel.
273
///    4. The last reference to the `Channel` is dropped. (Note that every circuit
274
///       on a `Channel` keeps a reference to it, which will in turn keep the
275
///       channel from closing until all those circuits have gone away.)
276
///
277
/// Note that in cases 1-3, the [`Channel`] object itself will still exist: it
278
/// will just be unusable for most purposes.  Most operations on it will fail
279
/// with an error.
280
pub struct Channel {
281
    /// A channel used to send control messages to the Reactor.
282
    control: mpsc::UnboundedSender<CtrlMsg>,
283
    /// A channel used to send cells to the Reactor.
284
    cell_tx: CellTx,
285

            
286
    /// A receiver that indicates whether the channel is closed.
287
    ///
288
    /// Awaiting will return a `CancelledError` event when the reactor is dropped.
289
    /// Read to decide if operations may succeed, and is returned by `wait_for_close`.
290
    reactor_closed_rx: oneshot_broadcast::Receiver<Result<CloseInfo>>,
291

            
292
    /// Padding controller, used to report when data is queued for this channel.
293
    padding_ctrl: PaddingController,
294

            
295
    /// A unique identifier for this channel.
296
    unique_id: UniqId,
297
    /// Target identity and address information for this peer.
298
    peer_id: OwnedChanTarget,
299
    /// Validated information for this peer.
300
    #[expect(unused)] // TODO(relay) Remove once used un choose_channel()
301
    peer: MaybeSensitive<PeerInfo>,
302
    /// The declared clock skew on this channel, at the time when this channel was
303
    /// created.
304
    clock_skew: ClockSkew,
305
    /// The time when this channel was successfully completed
306
    opened_at: coarsetime::Instant,
307
    /// Mutable state used by the `Channel.
308
    mutable: Mutex<MutableDetails>,
309
    /// Information shared with the reactor
310
    details: Arc<ChannelDetails>,
311
    /// Canonicity of this channel.
312
    canonicity: Canonicity,
313
}
314

            
315
/// This is information shared between the reactor and the frontend (`Channel` object).
316
///
317
/// `control` can't be here because we rely on it getting dropped when the last user goes away.
318
#[derive(Debug)]
319
pub(crate) struct ChannelDetails {
320
    /// Since when the channel became unused.
321
    ///
322
    /// If calling `time_since_update` returns None,
323
    /// this channel is still in use by at least one circuit.
324
    ///
325
    /// Set by reactor when a circuit is added or removed.
326
    /// Read from `Channel::duration_unused`.
327
    unused_since: AtomicOptTimestamp,
328
    /// Memory quota account
329
    ///
330
    /// This is here partly because we need to ensure it lives as long as the channel,
331
    /// as otherwise the memquota system will tear the account down.
332
    #[allow(dead_code)]
333
    memquota: ChannelAccount,
334
}
335

            
336
/// Mutable details (state) used by the `Channel` (frontend)
337
#[derive(Debug, Default)]
338
struct MutableDetails {
339
    /// State used to control padding
340
    padding: PaddingControlState,
341
}
342

            
343
/// State used to control padding
344
///
345
/// We store this here because:
346
///
347
///  1. It must be per-channel, because it depends on channel usage.  So it can't be in
348
///     (for example) `ChannelPaddingInstructionsUpdate`.
349
///
350
///  2. It could be in the channel manager's per-channel state but (for code flow reasons
351
///     there, really) at the point at which the channel manager concludes for a pending
352
///     channel that it ought to update the usage, it has relinquished the lock on its own data
353
///     structure.
354
///     And there is actually no need for this to be global: a per-channel lock is better than
355
///     reacquiring the global one.
356
///
357
///  3. It doesn't want to be in the channel reactor since that's super hot.
358
///
359
/// See also the overview at [`tor_proto::channel::padding`](padding)
360
#[derive(Debug, Educe)]
361
#[educe(Default)]
362
enum PaddingControlState {
363
    /// No usage of this channel, so far, implies sending or negotiating channel padding.
364
    ///
365
    /// This means we do not send (have not sent) any `ChannelPaddingInstructionsUpdates` to the reactor,
366
    /// with the following consequences:
367
    ///
368
    ///  * We don't enable our own padding.
369
    ///  * We don't do any work to change the timeout distribution in the padding timer,
370
    ///    (which is fine since this timer is not enabled).
371
    ///  * We don't send any PADDING_NEGOTIATE cells.  The peer is supposed to come to the
372
    ///    same conclusions as us, based on channel usage: it should also not send padding.
373
    #[educe(Default)]
374
    UsageDoesNotImplyPadding {
375
        /// The last padding parameters (from reparameterize)
376
        ///
377
        /// We keep this so that we can send it if and when
378
        /// this channel starts to be used in a way that implies (possibly) sending padding.
379
        padding_params: ChannelPaddingInstructionsUpdates,
380
    },
381

            
382
    /// Some usage of this channel implies possibly sending channel padding
383
    ///
384
    /// The required padding timer, negotiation cell, etc.,
385
    /// have been communicated to the reactor via a `CtrlMsg::ConfigUpdate`.
386
    ///
387
    /// Once we have set this variant, it remains this way forever for this channel,
388
    /// (the spec speaks of channels "only used for" certain purposes not getting padding).
389
    PaddingConfigured,
390
}
391

            
392
use PaddingControlState as PCS;
393

            
394
cfg_if! {
395
    if #[cfg(feature="circ-padding")] {
396
        /// Implementation type for a ChannelSender.
397
        type CellTx = CountingSink<mq_queue::Sender<ChanCellQueueEntry, mq_queue::MpscSpec>>;
398

            
399
        /// Implementation type for a cell queue held by a reactor.
400
        type CellRx = CountingStream<mq_queue::Receiver<ChanCellQueueEntry, mq_queue::MpscSpec>>;
401
    } else {
402
        /// Implementation type for a ChannelSender.
403
        type CellTx = mq_queue::Sender<ChanCellQueueEntry, mq_queue::MpscSpec>;
404

            
405
        /// Implementation type for a cell queue held by a reactor.
406
        type CellRx = mq_queue::Receiver<ChanCellQueueEntry, mq_queue::MpscSpec>;
407
    }
408
}
409

            
410
/// A handle to a [`Channel`]` that can be used, by circuits, to send channel cells.
411
#[derive(Debug)]
412
pub(crate) struct ChannelSender {
413
    /// MPSC sender to send cells.
414
    cell_tx: CellTx,
415
    /// A receiver used to check if the channel is closed.
416
    reactor_closed_rx: oneshot_broadcast::Receiver<Result<CloseInfo>>,
417
    /// Unique ID for this channel. For logging.
418
    unique_id: UniqId,
419
    /// Padding controller for this channel:
420
    /// used to report when we queue data that will eventually wind up on the channel.
421
    padding_ctrl: PaddingController,
422
}
423

            
424
impl ChannelSender {
425
    /// Check whether a cell type is permissible to be _sent_ on an
426
    /// open client channel.
427
4688
    fn check_cell(&self, cell: &AnyChanCell) -> Result<()> {
428
        use tor_cell::chancell::msg::AnyChanMsg::*;
429
4688
        let msg = cell.msg();
430
4688
        match msg {
431
12
            Created(_) | Created2(_) | CreatedFast(_) => Err(Error::from(internal!(
432
12
                "Can't send {} cell on client channel",
433
12
                msg.cmd()
434
12
            ))),
435
            Certs(_) | Versions(_) | Authenticate(_) | AuthChallenge(_) | Netinfo(_) => {
436
12
                Err(Error::from(internal!(
437
12
                    "Can't send {} cell after handshake is done",
438
12
                    msg.cmd()
439
12
                )))
440
            }
441
4664
            _ => Ok(()),
442
        }
443
4688
    }
444

            
445
    /// Obtain a reference to the `ChannelSender`'s [`DynTimeProvider`]
446
    ///
447
    /// (This can sometimes be used to avoid having to keep
448
    /// a separate clone of the time provider.)
449
72
    pub(crate) fn time_provider(&self) -> &DynTimeProvider {
450
        cfg_if! {
451
            if #[cfg(feature="circ-padding")] {
452
72
                self.cell_tx.inner().time_provider()
453
            } else {
454
                self.cell_tx.time_provider()
455
            }
456
        }
457
72
    }
458

            
459
    /// Return an approximate count of the number of outbound cells queued for this channel.
460
    ///
461
    /// This count is necessarily approximate,
462
    /// because the underlying count can be modified by other senders and receivers
463
    /// between when this method is called and when its return value is used.
464
    ///
465
    /// Does not include cells that have already been passed to the TLS connection.
466
    ///
467
    /// Circuit padding uses this count to determine
468
    /// when messages are already outbound for the first hop of a circuit.
469
    #[cfg(feature = "circ-padding")]
470
    pub(crate) fn approx_count(&self) -> usize {
471
        self.cell_tx.approx_count()
472
    }
473

            
474
    /// Note that a cell has been queued that will eventually be placed onto this sender.
475
    ///
476
    /// We use this as an input for padding machines.
477
4640
    pub(crate) fn note_cell_queued(&self) {
478
4640
        self.padding_ctrl.queued_data(crate::HopNum::from(0));
479
4640
    }
480
}
481

            
482
impl Sink<ChanCellQueueEntry> for ChannelSender {
483
    type Error = Error;
484

            
485
19018
    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
486
19018
        let this = self.get_mut();
487
19018
        Pin::new(&mut this.cell_tx)
488
19018
            .poll_ready(cx)
489
19019
            .map_err(|_| ChannelClosed.into())
490
19018
    }
491

            
492
4652
    fn start_send(self: Pin<&mut Self>, cell: ChanCellQueueEntry) -> Result<()> {
493
4652
        let this = self.get_mut();
494
4652
        if this.reactor_closed_rx.is_ready() {
495
            return Err(ChannelClosed.into());
496
4652
        }
497
4652
        this.check_cell(&cell.0)?;
498
        {
499
            use tor_cell::chancell::msg::AnyChanMsg::*;
500
4652
            match cell.0.msg() {
501
4524
                Relay(_) | Padding(_) | Vpadding(_) => {} // too frequent to log.
502
128
                _ => trace!(
503
                    channel_id = %this.unique_id,
504
                    "Sending {} for {}",
505
                    cell.0.msg().cmd(),
506
                    CircId::get_or_zero(cell.0.circid())
507
                ),
508
            }
509
        }
510

            
511
4652
        Pin::new(&mut this.cell_tx)
512
4652
            .start_send(cell)
513
4652
            .map_err(|_| ChannelClosed.into())
514
4652
    }
515

            
516
84
    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
517
84
        let this = self.get_mut();
518
84
        Pin::new(&mut this.cell_tx)
519
84
            .poll_flush(cx)
520
84
            .map_err(|_| ChannelClosed.into())
521
84
    }
522

            
523
    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
524
        let this = self.get_mut();
525
        Pin::new(&mut this.cell_tx)
526
            .poll_close(cx)
527
            .map_err(|_| ChannelClosed.into())
528
    }
529
}
530

            
531
impl Channel {
532
    /// Construct a channel and reactor.
533
    ///
534
    /// Internal method, called to finalize the channel when we've
535
    /// sent our netinfo cell, received the peer's netinfo cell, and
536
    /// we're finally ready to create circuits.
537
    ///
538
    /// Quick note on the allow clippy. This is has one call site so for now, it is fine that we
539
    /// bust the mighty 7 arguments.
540
    #[allow(clippy::too_many_arguments)] // TODO consider if we want a builder
541
496
    fn new<R>(
542
496
        channel_mode: ChannelMode,
543
496
        link_protocol: u16,
544
496
        sink: BoxedChannelSink,
545
496
        stream: BoxedChannelStream,
546
496
        streamops: BoxedChannelStreamOps,
547
496
        unique_id: UniqId,
548
496
        peer_id: OwnedChanTarget,
549
496
        peer: MaybeSensitive<PeerInfo>,
550
496
        clock_skew: ClockSkew,
551
496
        runtime: R,
552
496
        memquota: ChannelAccount,
553
496
        canonicity: Canonicity,
554
496
    ) -> Result<(Arc<Self>, reactor::Reactor<R>)>
555
496
    where
556
496
        R: Runtime,
557
    {
558
        use circmap::{CircIdRange, CircMap};
559
496
        let circid_range = match channel_mode {
560
            // client channels always originate here
561
496
            ChannelMode::Client => CircIdRange::High,
562
            #[cfg(feature = "relay")]
563
            ChannelMode::Relay { circ_id_range, .. } => circ_id_range,
564
        };
565
496
        let circmap = CircMap::new(circid_range);
566
496
        let dyn_time = DynTimeProvider::new(runtime.clone());
567

            
568
496
        let (control_tx, control_rx) = mpsc::unbounded();
569
496
        let (cell_tx, cell_rx) = mq_queue::MpscSpec::new(CHANNEL_BUFFER_SIZE)
570
496
            .new_mq(dyn_time.clone(), memquota.as_raw_account())?;
571
        #[cfg(feature = "circ-padding")]
572
496
        let (cell_tx, cell_rx) = counting_streams::channel(cell_tx, cell_rx);
573
496
        let unused_since = AtomicOptTimestamp::new();
574
496
        unused_since.update();
575

            
576
496
        let mutable = MutableDetails::default();
577
496
        let (reactor_closed_tx, reactor_closed_rx) = oneshot_broadcast::channel();
578

            
579
496
        let details = ChannelDetails {
580
496
            unused_since,
581
496
            memquota,
582
496
        };
583
496
        let details = Arc::new(details);
584

            
585
        // We might be using experimental maybenot padding; this creates the padding framework for that.
586
        //
587
        // TODO: This backend is currently optimized for circuit padding,
588
        // so it might allocate a bit more than necessary to account for multiple hops.
589
        // We should tune it when we deploy padding in production.
590
496
        let (padding_ctrl, padding_event_stream) =
591
496
            client::circuit::padding::new_padding(DynTimeProvider::new(runtime.clone()));
592

            
593
496
        let channel = Arc::new(Channel {
594
496
            control: control_tx,
595
496
            cell_tx,
596
496
            reactor_closed_rx,
597
496
            padding_ctrl: padding_ctrl.clone(),
598
496
            unique_id,
599
496
            peer_id,
600
496
            peer,
601
496
            clock_skew,
602
496
            opened_at: coarsetime::Instant::now(),
603
496
            mutable: Mutex::new(mutable),
604
496
            details: Arc::clone(&details),
605
496
            canonicity,
606
496
        });
607

            
608
        // We start disabled; the channel manager will `reconfigure` us soon after creation.
609
496
        let padding_timer = Box::pin(padding::Timer::new_disabled(runtime.clone(), None)?);
610

            
611
        cfg_if! {
612
            if #[cfg(feature = "circ-padding")] {
613
                use crate::util::sink_blocker::{SinkBlocker,CountingPolicy};
614
496
                let sink = SinkBlocker::new(sink, CountingPolicy::new_unlimited());
615
            }
616
        }
617

            
618
        #[cfg(feature = "relay")]
619
496
        let create_request_handler: Option<_> = match channel_mode {
620
            ChannelMode::Relay {
621
                create_request_handler,
622
                our_ed25519_id,
623
                our_rsa_id,
624
                ..
625
            } => Some(CreateRequestHandlerAndData {
626
                handler: create_request_handler,
627
                channel: Arc::downgrade(&channel),
628
                our_ed25519_id,
629
                our_rsa_id,
630
            }),
631
496
            ChannelMode::Client => None,
632
        };
633
        // clippy wants us to consume `channel_mode` (`needless_pass_by_value`)
634
        #[cfg(not(feature = "relay"))]
635
        #[expect(clippy::drop_non_drop)]
636
        drop(channel_mode);
637

            
638
496
        let reactor = Reactor {
639
496
            runtime,
640
496
            control: control_rx,
641
496
            cells: cell_rx,
642
496
            reactor_closed_tx,
643
496
            input: futures::StreamExt::fuse(stream),
644
496
            output: sink,
645
496
            streamops,
646
496
            circs: circmap,
647
496
            circ_unique_id_ctx: CircUniqIdContext::new(),
648
496
            link_protocol,
649
496
            unique_id,
650
496
            details,
651
496
            #[cfg(feature = "relay")]
652
496
            create_request_handler,
653
496
            padding_timer,
654
496
            padding_ctrl,
655
496
            padding_event_stream,
656
496
            padding_blocker: None,
657
496
            special_outgoing: Default::default(),
658
496
        };
659

            
660
496
        Ok((channel, reactor))
661
496
    }
662

            
663
    /// Return a process-unique identifier for this channel.
664
4
    pub fn unique_id(&self) -> UniqId {
665
4
        self.unique_id
666
4
    }
667

            
668
    /// Return a reference to the memory tracking account for this Channel
669
    pub fn mq_account(&self) -> &ChannelAccount {
670
        &self.details.memquota
671
    }
672

            
673
    /// Obtain a reference to the `Channel`'s [`DynTimeProvider`]
674
    ///
675
    /// (This can sometimes be used to avoid having to keep
676
    /// a separate clone of the time provider.)
677
392
    pub fn time_provider(&self) -> &DynTimeProvider {
678
        cfg_if! {
679
            if #[cfg(feature="circ-padding")] {
680
392
                self.cell_tx.inner().time_provider()
681
            } else {
682
                self.cell_tx.time_provider()
683
            }
684
        }
685
392
    }
686

            
687
    /// Return an OwnedChanTarget representing the actual handshake used to
688
    /// create this channel.
689
1200
    pub fn target(&self) -> &OwnedChanTarget {
690
1200
        &self.peer_id
691
1200
    }
692

            
693
    /// Return the amount of time that has passed since this channel became open.
694
    pub fn age(&self) -> Duration {
695
        self.opened_at.elapsed().into()
696
    }
697

            
698
    /// Return a ClockSkew declaring how much clock skew the other side of this channel
699
    /// claimed that we had when we negotiated the connection.
700
    pub fn clock_skew(&self) -> ClockSkew {
701
        self.clock_skew
702
    }
703

            
704
    /// Send a control message
705
    #[instrument(level = "trace", skip_all)]
706
2988
    fn send_control(&self, msg: CtrlMsg) -> StdResult<(), ChannelClosed> {
707
2988
        self.control
708
2988
            .unbounded_send(msg)
709
2988
            .map_err(|_| ChannelClosed)?;
710
2900
        Ok(())
711
2988
    }
712

            
713
    /// Acquire the lock on `mutable` (and handle any poison error)
714
1152
    fn mutable(&self) -> MutexGuard<MutableDetails> {
715
1152
        self.mutable.lock().expect("channel details poisoned")
716
1152
    }
717

            
718
    /// Specify that this channel should do activities related to channel padding
719
    ///
720
    /// Initially, the channel does nothing related to channel padding:
721
    /// it neither sends any padding, nor sends any PADDING_NEGOTIATE cells.
722
    ///
723
    /// After this function has been called, it will do both,
724
    /// according to the parameters specified through `reparameterize`.
725
    /// Note that this might include *disabling* padding
726
    /// (for example, by sending a `PADDING_NEGOTIATE`).
727
    ///
728
    /// Idempotent.
729
    ///
730
    /// There is no way to undo the effect of this call.
731
    #[instrument(level = "trace", skip_all)]
732
1152
    pub fn engage_padding_activities(&self) {
733
1152
        let mut mutable = self.mutable();
734

            
735
1152
        match &mutable.padding {
736
            PCS::UsageDoesNotImplyPadding {
737
1152
                padding_params: params,
738
            } => {
739
                // Well, apparently the channel usage *does* imply padding now,
740
                // so we need to (belatedly) enable the timer,
741
                // send the padding negotiation cell, etc.
742
1152
                let mut params = params.clone();
743

            
744
                // Except, maybe the padding we would be requesting is precisely default,
745
                // so we wouldn't actually want to send that cell.
746
1152
                if params.padding_negotiate == Some(PaddingNegotiate::start_default()) {
747
                    params.padding_negotiate = None;
748
1152
                }
749

            
750
1152
                match self.send_control(CtrlMsg::ConfigUpdate(Arc::new(params))) {
751
1152
                    Ok(()) => {}
752
                    Err(ChannelClosed) => return,
753
                }
754

            
755
1152
                mutable.padding = PCS::PaddingConfigured;
756
            }
757

            
758
            PCS::PaddingConfigured => {
759
                // OK, nothing to do
760
            }
761
        }
762

            
763
1152
        drop(mutable); // release the lock now: lock span covers the send, ensuring ordering
764
1152
    }
765

            
766
    /// Reparameterise (update parameters; reconfigure)
767
    ///
768
    /// Returns `Err` if the channel was closed earlier
769
    #[instrument(level = "trace", skip_all)]
770
2592
    pub fn reparameterize(&self, params: Arc<ChannelPaddingInstructionsUpdates>) -> Result<()> {
771
2592
        let mut mutable = self
772
2592
            .mutable
773
2592
            .lock()
774
2592
            .map_err(|_| internal!("channel details poisoned"))?;
775

            
776
2592
        match &mut mutable.padding {
777
            PCS::PaddingConfigured => {
778
1440
                self.send_control(CtrlMsg::ConfigUpdate(params))?;
779
            }
780
1152
            PCS::UsageDoesNotImplyPadding { padding_params } => {
781
1152
                padding_params.combine(&params);
782
1152
            }
783
        }
784

            
785
2592
        drop(mutable); // release the lock now: lock span covers the send, ensuring ordering
786
2592
        Ok(())
787
2592
    }
788

            
789
    /// Update the KIST parameters.
790
    ///
791
    /// Returns `Err` if the channel is closed.
792
    #[instrument(level = "trace", skip_all)]
793
    pub fn reparameterize_kist(&self, kist_params: KistParams) -> Result<()> {
794
        Ok(self.send_control(CtrlMsg::KistConfigUpdate(kist_params))?)
795
    }
796

            
797
    /// Return an error if this channel is somehow mismatched with the
798
    /// given target.
799
42
    pub fn check_match<T: HasRelayIds + ?Sized>(&self, target: &T) -> Result<()> {
800
42
        check_id_match_helper(&self.peer_id, target)
801
42
    }
802

            
803
    /// Return true if this channel is closed and therefore unusable.
804
124
    pub fn is_closing(&self) -> bool {
805
124
        self.reactor_closed_rx.is_ready()
806
124
    }
807

            
808
    /// Return true iff this channel is considered canonical by us.
809
    pub fn is_canonical(&self) -> bool {
810
        self.canonicity.peer_is_canonical
811
    }
812

            
813
    /// Return true if we think the peer considers this channel as canonical.
814
    pub fn is_canonical_to_peer(&self) -> bool {
815
        self.canonicity.canonical_to_peer
816
    }
817

            
818
    /// If the channel is not in use, return the amount of time
819
    /// it has had with no circuits.
820
    ///
821
    /// Return `None` if the channel is currently in use.
822
182
    pub fn duration_unused(&self) -> Option<std::time::Duration> {
823
182
        self.details
824
182
            .unused_since
825
182
            .time_since_update()
826
182
            .map(Into::into)
827
182
    }
828

            
829
    /// Return a new [`ChannelSender`] to transmit cells on this channel.
830
432
    pub(crate) fn sender(&self) -> ChannelSender {
831
432
        ChannelSender {
832
432
            cell_tx: self.cell_tx.clone(),
833
432
            reactor_closed_rx: self.reactor_closed_rx.clone(),
834
432
            unique_id: self.unique_id,
835
432
            padding_ctrl: self.padding_ctrl.clone(),
836
432
        }
837
432
    }
838

            
839
    /// Return a newly allocated PendingClientTunnel object with
840
    /// a corresponding tunnel reactor. A circuit ID is allocated, but no
841
    /// messages are sent, and no cryptography is done.
842
    ///
843
    /// To use the results of this method, call Reactor::run() in a
844
    /// new task, then use the methods of
845
    /// [crate::client::circuit::PendingClientTunnel] to build the circuit.
846
    #[instrument(level = "trace", skip_all)]
847
12
    pub async fn new_tunnel(
848
12
        self: &Arc<Self>,
849
12
        timeouts: Arc<dyn TimeoutEstimator>,
850
18
    ) -> Result<(PendingClientTunnel, client::reactor::Reactor)> {
851
        if self.is_closing() {
852
            return Err(ChannelClosed.into());
853
        }
854

            
855
        let time_prov = self.time_provider().clone();
856
        let memquota = CircuitAccount::new(&self.details.memquota)?;
857

            
858
        // TODO: blocking is risky, but so is unbounded.
859
        let (sender, receiver) =
860
            MpscSpec::new(128).new_mq(time_prov.clone(), memquota.as_raw_account())?;
861
        let (createdsender, createdreceiver) = oneshot::channel::<CreateResponse>();
862

            
863
        let (tx, rx) = oneshot::channel();
864
        self.send_control(CtrlMsg::AllocateCircuit {
865
            created_sender: createdsender,
866
            sender,
867
            tx,
868
        })?;
869
        let (id, circ_unique_id, padding_ctrl, padding_stream) =
870
            rx.await.map_err(|_| ChannelClosed)??;
871

            
872
        trace!("{}: Allocated CircId {}", circ_unique_id, id);
873

            
874
        Ok(PendingClientTunnel::new(
875
            id,
876
            self.clone(),
877
            createdreceiver,
878
            receiver,
879
            circ_unique_id,
880
            time_prov,
881
            memquota,
882
            padding_ctrl,
883
            padding_stream,
884
            timeouts,
885
        ))
886
12
    }
887

            
888
    /// Return a newly allocated outbound relay circuit with.
889
    ///
890
    /// A circuit ID is allocated, but no messages are sent, and no cryptography is done.
891
    ///
892
    // TODO(relay): this duplicates much of new_tunnel above, but I expect
893
    // the implementations to diverge once we introduce a new CtrlMsg for
894
    // allocating relay circuits.
895
    #[cfg(feature = "relay")]
896
4
    pub(crate) async fn new_outbound_circ(
897
4
        self: &Arc<Self>,
898
4
        memquota: CircuitAccount,
899
6
    ) -> Result<(CircId, CircuitRxReceiver, oneshot::Receiver<CreateResponse>)> {
900
4
        if self.is_closing() {
901
            return Err(ChannelClosed.into());
902
4
        }
903

            
904
4
        let time_prov = self.time_provider().clone();
905

            
906
        // TODO: blocking is risky, but so is unbounded.
907
4
        let (sender, receiver) =
908
4
            MpscSpec::new(128).new_mq(time_prov.clone(), memquota.as_raw_account())?;
909
4
        let (createdsender, createdreceiver) = oneshot::channel::<CreateResponse>();
910

            
911
4
        let (tx, rx) = oneshot::channel();
912

            
913
4
        self.send_control(CtrlMsg::AllocateCircuit {
914
4
            created_sender: createdsender,
915
4
            sender,
916
4
            tx,
917
4
        })?;
918

            
919
        // TODO(relay): I don't think we need circuit-level padding on this side of the circuit.
920
        // This just drops the padding controller and corresponding event stream,
921
        // but maybe it would be better to just not set it up in the first place?
922
        // This suggests we might need a different control command for allocating
923
        // the outbound relay circuits...
924
4
        let (id, circ_unique_id, _padding_ctrl, _padding_stream) =
925
4
            rx.await.map_err(|_| ChannelClosed)??;
926

            
927
4
        let channel_account = self.details.memquota.as_raw_account();
928
        // Link the memquota circuit account with the outbound channel account:
929
4
        memquota.as_raw_account().add_parent(channel_account)?;
930

            
931
4
        trace!("{}: Allocated CircId {}", circ_unique_id, id);
932

            
933
4
        Ok((id, receiver, createdreceiver))
934
4
    }
935

            
936
    /// Shut down this channel immediately, along with all circuits that
937
    /// are using it.
938
    ///
939
    /// Note that other references to this channel may exist.  If they
940
    /// do, they will stop working after you call this function.
941
    ///
942
    /// It's not necessary to call this method if you're just done
943
    /// with a channel: the channel should close on its own once nothing
944
    /// is using it any more.
945
    #[instrument(level = "trace", skip_all)]
946
36
    pub fn terminate(&self) {
947
36
        let _ = self.send_control(CtrlMsg::Shutdown);
948
36
    }
949

            
950
    /// Tell the reactor that the circuit with the given ID has gone away.
951
    #[instrument(level = "trace", skip_all)]
952
344
    pub fn close_circuit(&self, circid: CircId) -> Result<()> {
953
344
        self.send_control(CtrlMsg::CloseCircuit(circid))?;
954
256
        Ok(())
955
344
    }
956

            
957
    /// Return a future that will resolve once this channel has closed.
958
    ///
959
    /// Note that this method does not _cause_ the channel to shut down on its own.
960
36
    pub fn wait_for_close(
961
36
        &self,
962
36
    ) -> impl Future<Output = StdResult<CloseInfo, ClosedUnexpectedly>> + Send + Sync + 'static + use<>
963
    {
964
36
        self.reactor_closed_rx
965
36
            .clone()
966
36
            .into_future()
967
48
            .map(|recv| match recv {
968
12
                Ok(Ok(info)) => Ok(info),
969
12
                Ok(Err(e)) => Err(ClosedUnexpectedly::ReactorError(e)),
970
12
                Err(oneshot_broadcast::SenderDropped) => Err(ClosedUnexpectedly::ReactorDropped),
971
36
            })
972
36
    }
973

            
974
    /// Install a [`CircuitPadder`](client::CircuitPadder) for this channel.
975
    ///
976
    /// Replaces any previous padder installed.
977
    #[cfg(feature = "circ-padding-manual")]
978
    pub async fn start_padding(self: &Arc<Self>, padder: client::CircuitPadder) -> Result<()> {
979
        self.set_padder_impl(Some(padder)).await
980
    }
981

            
982
    /// Remove any [`CircuitPadder`](client::CircuitPadder) installed for this channel.
983
    ///
984
    /// Does nothing if there was not a padder installed there.
985
    #[cfg(feature = "circ-padding-manual")]
986
    pub async fn stop_padding(self: &Arc<Self>) -> Result<()> {
987
        self.set_padder_impl(None).await
988
    }
989

            
990
    /// Replace the [`CircuitPadder`](client::CircuitPadder) installed for this channel with `padder`.
991
    #[cfg(feature = "circ-padding-manual")]
992
    async fn set_padder_impl(
993
        self: &Arc<Self>,
994
        padder: Option<client::CircuitPadder>,
995
    ) -> Result<()> {
996
        let (tx, rx) = oneshot::channel();
997
        let msg = CtrlMsg::SetChannelPadder { padder, sender: tx };
998
        self.control
999
            .unbounded_send(msg)
            .map_err(|_| Error::ChannelClosed(ChannelClosed))?;
        rx.await.map_err(|_| Error::ChannelClosed(ChannelClosed))?
    }
    /// Make a new fake reactor-less channel.  For testing only, obviously.
    ///
    /// Returns the receiver end of the control message mpsc.
    ///
    /// Suitable for external callers who want to test behaviour
    /// of layers including the logic in the channel frontend
    /// (`Channel` object methods).
    //
    // This differs from test::fake_channel as follows:
    //  * It returns the mpsc Receiver
    //  * It does not require explicit specification of details
    #[cfg(feature = "testing")]
48
    pub fn new_fake(
48
        rt: impl SleepProvider + CoarseTimeProvider,
48
        _channel_type: ChannelType,
48
    ) -> (Channel, mpsc::UnboundedReceiver<CtrlMsg>) {
48
        let (control, control_recv) = mpsc::unbounded();
48
        let details = fake_channel_details();
48
        let unique_id = UniqId::new();
48
        let peer_id = OwnedChanTarget::builder()
48
            .ed_identity([6_u8; 32].into())
48
            .rsa_identity([10_u8; 20].into())
48
            .build()
48
            .expect("Couldn't construct peer id");
        // This will make rx trigger immediately.
48
        let (_tx, rx) = oneshot_broadcast::channel();
48
        let (padding_ctrl, _) = client::circuit::padding::new_padding(DynTimeProvider::new(rt));
48
        let channel = Channel {
48
            control,
48
            cell_tx: fake_mpsc().0,
48
            reactor_closed_rx: rx,
48
            padding_ctrl,
48
            unique_id,
48
            peer_id,
48
            peer: MaybeSensitive::not_sensitive(PeerInfo::EMPTY),
48
            clock_skew: ClockSkew::None,
48
            opened_at: coarsetime::Instant::now(),
48
            mutable: Default::default(),
48
            details,
48
            canonicity: Canonicity::new_canonical(),
48
        };
48
        (channel, control_recv)
48
    }
}
/// If there is any identity in `wanted_ident` that is not present in
/// `my_ident`, return a ChanMismatch error.
///
/// This is a helper for [`Channel::check_match`] and
/// UnverifiedChannel::check_internal.
56
fn check_id_match_helper<T, U>(my_ident: &T, wanted_ident: &U) -> Result<()>
56
where
56
    T: HasRelayIds + ?Sized,
56
    U: HasRelayIds + ?Sized,
{
82
    for desired in wanted_ident.identities() {
82
        let id_type = desired.id_type();
82
        match my_ident.identity(id_type) {
82
            Some(actual) if actual == desired => {}
8
            Some(actual) => {
8
                return Err(Error::ChanMismatch(format!(
8
                    "Identity {} does not match target {}",
8
                    sv(actual),
8
                    sv(desired)
8
                )));
            }
            None => {
                return Err(Error::ChanMismatch(format!(
                    "Peer does not have {} identity",
                    id_type
                )));
            }
        }
    }
48
    Ok(())
56
}
impl HasRelayIds for Channel {
4656
    fn identity(
4656
        &self,
4656
        key_type: tor_linkspec::RelayIdType,
4656
    ) -> Option<tor_linkspec::RelayIdRef<'_>> {
4656
        self.peer_id.identity(key_type)
4656
    }
}
/// The status of a channel which was closed successfully.
///
/// **Note:** This doesn't have any associated data,
/// but may be expanded in the future.
// I can't think of any info we'd want to return to waiters,
// but this type leaves the possibility open without requiring any backwards-incompatible changes.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct CloseInfo;
/// The status of a channel which closed unexpectedly.
#[derive(Clone, Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ClosedUnexpectedly {
    /// The channel reactor was dropped or panicked before completing.
    #[error("channel reactor was dropped or panicked before completing")]
    ReactorDropped,
    /// The channel reactor had an internal error.
    #[error("channel reactor had an internal error")]
    ReactorError(Error),
}
/// Whether the channel is operating in "client" or "relay" mode,
/// and some mode-specific parameters.
pub(crate) enum ChannelMode {
    /// An incoming channel,
    /// or an outgoing channel made by a non-bridge relay.
    #[cfg(feature = "relay")]
    Relay {
        /// A handler for CREATE2/CREATE_FAST messages.
        create_request_handler: Arc<CreateRequestHandler>,
        /// Our Ed25519 identity.
        our_ed25519_id: Ed25519Identity,
        /// Our RSA identity.
        our_rsa_id: RsaIdentity,
        /// The range of circuit IDs that we allocate for new circuits.
        circ_id_range: circmap::CircIdRange,
    },
    /// An outgoing channel made by a client or bridge relay.
    Client,
}
impl ChannelMode {
    /// Returns an error if the mode doesn't agree with the channel type.
50
    pub(crate) fn check_agrees_with_type(
50
        &self,
50
        channel_type: ChannelType,
50
    ) -> StdResult<(), tor_error::Bug> {
        use ChannelType::*;
        use circmap::CircIdRange::*;
50
        match (channel_type, self) {
50
            (ClientInitiator, Self::Client) => {}
            #[cfg(feature = "relay")]
            #[rustfmt::skip]
            (RelayInitiator, Self::Relay { circ_id_range: High, .. }) => {}
            #[cfg(feature = "relay")]
            #[rustfmt::skip]
            (RelayResponder { .. }, Self::Relay { circ_id_range: Low, .. }) => {}
            _ => return Err(internal!("`ChannelMode` doesn't agree with `ChannelType`")),
        }
50
        Ok(())
50
    }
}
/// Make some fake channel details (for testing only!)
#[cfg(any(test, feature = "testing"))]
1174
fn fake_channel_details() -> Arc<ChannelDetails> {
1174
    let unused_since = AtomicOptTimestamp::new();
1174
    Arc::new(ChannelDetails {
1174
        unused_since,
1174
        memquota: crate::util::fake_mq(),
1174
    })
1174
}
/// Make an MPSC queue, of the type we use in Channels, but a fake one for testing
#[cfg(any(test, feature = "testing"))] // Used by Channel::new_fake which is also feature=testing
1172
pub(crate) fn fake_mpsc() -> (CellTx, CellRx) {
1172
    let (tx, rx) = crate::fake_mpsc(CHANNEL_BUFFER_SIZE);
    #[cfg(feature = "circ-padding")]
1172
    let (tx, rx) = counting_streams::channel(tx, rx);
1172
    (tx, rx)
1172
}
#[cfg(test)]
pub(crate) mod test {
    // Most of this module is tested via tests that also check on the
    // reactor code; there are just a few more cases to examine here.
    #![allow(clippy::unwrap_used)]
    use super::*;
    pub(crate) use crate::channel::reactor::test::{CodecResult, new_reactor};
    use tor_cell::chancell::msg::HandshakeType;
    use tor_cell::chancell::{AnyChanCell, msg};
    use tor_rtcompat::test_with_one_runtime;
    /// Make a new fake reactor-less channel.  For testing only, obviously.
    pub(crate) fn fake_channel(
        rt: impl SleepProvider + CoarseTimeProvider,
        _channel_type: ChannelType,
    ) -> Channel {
        let unique_id = UniqId::new();
        let peer_id = OwnedChanTarget::builder()
            .ed_identity([6_u8; 32].into())
            .rsa_identity([10_u8; 20].into())
            .build()
            .expect("Couldn't construct peer id");
        // This will make rx trigger immediately.
        let (_tx, rx) = oneshot_broadcast::channel();
        let (padding_ctrl, _) = client::circuit::padding::new_padding(DynTimeProvider::new(rt));
        Channel {
            control: mpsc::unbounded().0,
            cell_tx: fake_mpsc().0,
            reactor_closed_rx: rx,
            padding_ctrl,
            unique_id,
            peer_id,
            peer: MaybeSensitive::not_sensitive(PeerInfo::EMPTY),
            clock_skew: ClockSkew::None,
            opened_at: coarsetime::Instant::now(),
            mutable: Default::default(),
            details: fake_channel_details(),
            canonicity: Canonicity::new_canonical(),
        }
    }
    #[test]
    fn send_bad() {
        tor_rtcompat::test_with_all_runtimes!(|rt| async move {
            use std::error::Error;
            let chan = fake_channel(rt, ChannelType::ClientInitiator);
            let cell = AnyChanCell::new(CircId::new(7), msg::Created2::new(&b"hihi"[..]).into());
            let e = chan.sender().check_cell(&cell);
            assert!(e.is_err());
            assert!(
                format!("{}", e.unwrap_err().source().unwrap())
                    .contains("Can't send CREATED2 cell on client channel")
            );
            let cell = AnyChanCell::new(None, msg::Certs::new_empty().into());
            let e = chan.sender().check_cell(&cell);
            assert!(e.is_err());
            assert!(
                format!("{}", e.unwrap_err().source().unwrap())
                    .contains("Can't send CERTS cell after handshake is done")
            );
            let cell = AnyChanCell::new(
                CircId::new(5),
                msg::Create2::new(HandshakeType::NTOR, &b"abc"[..]).into(),
            );
            let e = chan.sender().check_cell(&cell);
            assert!(e.is_ok());
            // FIXME(eta): more difficult to test that sending works now that it has to go via reactor
            // let got = output.next().await.unwrap();
            // assert!(matches!(got.msg(), ChanMsg::Create2(_)));
        });
    }
    #[test]
    fn check_match() {
        test_with_one_runtime!(|rt| async move {
            let chan = fake_channel(rt, ChannelType::ClientInitiator);
            let t1 = OwnedChanTarget::builder()
                .ed_identity([6; 32].into())
                .rsa_identity([10; 20].into())
                .build()
                .unwrap();
            let t2 = OwnedChanTarget::builder()
                .ed_identity([1; 32].into())
                .rsa_identity([3; 20].into())
                .build()
                .unwrap();
            let t3 = OwnedChanTarget::builder()
                .ed_identity([3; 32].into())
                .rsa_identity([2; 20].into())
                .build()
                .unwrap();
            assert!(chan.check_match(&t1).is_ok());
            assert!(chan.check_match(&t2).is_err());
            assert!(chan.check_match(&t3).is_err());
        });
    }
    #[test]
    fn unique_id() {
        test_with_one_runtime!(|rt| async move {
            let ch1 = fake_channel(rt.clone(), ChannelType::ClientInitiator);
            let ch2 = fake_channel(rt, ChannelType::ClientInitiator);
            assert_ne!(ch1.unique_id(), ch2.unique_id());
        });
    }
    #[test]
    fn duration_unused_at() {
        test_with_one_runtime!(|rt| async move {
            let details = fake_channel_details();
            let mut ch = fake_channel(rt, ChannelType::ClientInitiator);
            ch.details = details.clone();
            details.unused_since.update();
            assert!(ch.duration_unused().is_some());
        });
    }
}