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
#[cfg(test)]
64
pub(crate) mod test_utils;
65

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

            
95
#[cfg(feature = "circ-padding")]
96
use tor_async_utils::counting_streams::{self, CountingSink, CountingStream};
97

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

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

            
119
use asynchronous_codec;
120
use futures::channel::mpsc;
121
use futures::io::{AsyncRead, AsyncWrite};
122
use oneshot_fused_workaround as oneshot;
123

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

            
130
use tracing::{instrument, trace};
131

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

            
139
use kist::KistParams;
140

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

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

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

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

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

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

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

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

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

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

            
295
    /// Padding controller, used to report when data is queued for this channel.
296
    padding_ctrl: PaddingController,
297

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

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

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

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

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

            
394
use PaddingControlState as PCS;
395

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

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

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

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

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

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

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

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

            
484
impl Sink<ChanCellQueueEntry> for ChannelSender {
485
    type Error = Error;
486

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

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

            
513
4708
        Pin::new(&mut this.cell_tx)
514
4708
            .start_send(cell)
515
4708
            .map_err(|_| ChannelClosed.into())
516
4708
    }
517

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

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

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

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

            
578
568
        let mutable = MutableDetails::default();
579
568
        let (reactor_closed_tx, reactor_closed_rx) = oneshot_broadcast::channel();
580

            
581
568
        let details = ChannelDetails {
582
568
            unused_since,
583
568
            memquota,
584
568
        };
585
568
        let details = Arc::new(details);
586

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

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

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

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

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

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

            
662
568
        Ok((channel, reactor))
663
568
    }
664

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

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

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

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

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

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

            
706
    /// Send a control message
707
    #[instrument(level = "trace", skip_all)]
708
568
    #[cfg_attr(test, visibility::make(pub(crate)))]
709
3322
    fn send_control(&self, msg: CtrlMsg) -> StdResult<(), ChannelClosed> {
710
3322
        self.control
711
3322
            .unbounded_send(msg)
712
3322
            .map_err(|_| ChannelClosed)?;
713
3214
        Ok(())
714
3322
    }
715

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

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

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

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

            
753
1224
                match self.send_control(CtrlMsg::ConfigUpdate(Arc::new(params))) {
754
1224
                    Ok(()) => {}
755
                    Err(ChannelClosed) => return,
756
                }
757

            
758
1224
                mutable.padding = PCS::PaddingConfigured;
759
            }
760

            
761
            PCS::PaddingConfigured => {
762
                // OK, nothing to do
763
            }
764
        }
765

            
766
1224
        drop(mutable); // release the lock now: lock span covers the send, ensuring ordering
767
1224
    }
768

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

            
779
2754
        match &mut mutable.padding {
780
            PCS::PaddingConfigured => {
781
1530
                self.send_control(CtrlMsg::ConfigUpdate(params))?;
782
            }
783
1224
            PCS::UsageDoesNotImplyPadding { padding_params } => {
784
1224
                padding_params.combine(&params);
785
1224
            }
786
        }
787

            
788
2754
        drop(mutable); // release the lock now: lock span covers the send, ensuring ordering
789
2754
        Ok(())
790
2754
    }
791

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

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

            
806
    /// Return true if this channel is closed and therefore unusable.
807
165
    pub fn is_closing(&self) -> bool {
808
165
        self.reactor_closed_rx.is_ready()
809
165
    }
810

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

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

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

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

            
842
    /// Return the [`PeerInfo`] of this channel.
843
    #[cfg(feature = "relay")]
844
64
    pub(crate) fn peer_info(&self) -> &Arc<PeerInfo> {
845
64
        &self.peer
846
64
    }
847

            
848
    /// Return a newly allocated PendingClientTunnel object with
849
    /// a corresponding tunnel reactor. A circuit ID is allocated, but no
850
    /// messages are sent, and no cryptography is done.
851
    ///
852
    /// To use the results of this method, call Reactor::run() in a
853
    /// new task, then use the methods of
854
    /// [crate::client::circuit::PendingClientTunnel] to build the circuit.
855
    #[instrument(level = "trace", skip_all)]
856
42
    pub async fn new_tunnel(
857
42
        self: &Arc<Self>,
858
42
        timeouts: Arc<dyn TimeoutEstimator>,
859
63
    ) -> Result<(PendingClientTunnel, client::reactor::Reactor)> {
860
        if self.is_closing() {
861
            return Err(ChannelClosed.into());
862
        }
863

            
864
        let time_prov = self.time_provider().clone();
865
        let memquota = CircuitAccount::new(&self.details.memquota)?;
866

            
867
        // TODO: blocking is risky, but so is unbounded.
868
        let (sender, receiver) =
869
            MpscSpec::new(128).new_mq(time_prov.clone(), memquota.as_raw_account())?;
870
        let (sender, receiver) = crate::circuit::circ_sender::channel(sender, receiver);
871
        let (createdsender, createdreceiver) = oneshot::channel::<CreateResponse>();
872

            
873
        let (tx, rx) = oneshot::channel();
874

            
875
        self.send_control(CtrlMsg::AllocateCircuit {
876
            created_sender: createdsender,
877
            sender,
878
            tx,
879
        })?;
880
        let (circ_id, circ_unique_id, padding_ctrl, padding_stream) =
881
            rx.await.map_err(|_| ChannelClosed)??;
882

            
883
        trace!("{}: Allocated CircId {}", circ_unique_id, circ_id);
884

            
885
        Ok(PendingClientTunnel::new(
886
            circ_id,
887
            self.clone(),
888
            createdreceiver,
889
            receiver,
890
            circ_unique_id,
891
            time_prov,
892
            memquota,
893
            padding_ctrl,
894
            padding_stream,
895
            timeouts,
896
        ))
897
42
    }
898

            
899
    /// Return a newly allocated outbound relay circuit with.
900
    ///
901
    /// A circuit ID is allocated, but no messages are sent, and no cryptography is done.
902
    ///
903
    // TODO(relay): this duplicates much of new_tunnel above, but I expect
904
    // the implementations to diverge once we introduce a new CtrlMsg for
905
    // allocating relay circuits.
906
    #[cfg(feature = "relay")]
907
12
    pub(crate) async fn new_outbound_circ(
908
12
        self: &Arc<Self>,
909
12
        memquota: CircuitAccount,
910
18
    ) -> Result<(CircId, CircuitRxReceiver, oneshot::Receiver<CreateResponse>)> {
911
12
        if self.is_closing() {
912
            return Err(ChannelClosed.into());
913
12
        }
914

            
915
12
        let time_prov = self.time_provider().clone();
916

            
917
        // TODO: blocking is risky, but so is unbounded.
918
12
        let (sender, receiver) =
919
12
            MpscSpec::new(128).new_mq(time_prov.clone(), memquota.as_raw_account())?;
920
12
        let (sender, receiver) = crate::circuit::circ_sender::channel(sender, receiver);
921
12
        let (createdsender, createdreceiver) = oneshot::channel::<CreateResponse>();
922

            
923
12
        let (tx, rx) = oneshot::channel();
924

            
925
12
        self.send_control(CtrlMsg::AllocateCircuit {
926
12
            created_sender: createdsender,
927
12
            sender,
928
12
            tx,
929
12
        })?;
930

            
931
        // TODO(relay): I don't think we need circuit-level padding on this side of the circuit.
932
        // This just drops the padding controller and corresponding event stream,
933
        // but maybe it would be better to just not set it up in the first place?
934
        // This suggests we might need a different control command for allocating
935
        // the outbound relay circuits...
936
12
        let (id, circ_unique_id, _padding_ctrl, _padding_stream) =
937
12
            rx.await.map_err(|_| ChannelClosed)??;
938

            
939
12
        let channel_account = self.details.memquota.as_raw_account();
940
        // Link the memquota circuit account with the outbound channel account:
941
12
        memquota.as_raw_account().add_parent(channel_account)?;
942

            
943
12
        trace!("{}: Allocated CircId {}", circ_unique_id, id);
944

            
945
12
        Ok((id, receiver, createdreceiver))
946
12
    }
947

            
948
    /// Shut down this channel immediately, along with all circuits that
949
    /// are using it.
950
    ///
951
    /// Note that other references to this channel may exist.  If they
952
    /// do, they will stop working after you call this function.
953
    ///
954
    /// It's not necessary to call this method if you're just done
955
    /// with a channel: the channel should close on its own once nothing
956
    /// is using it any more.
957
    #[instrument(level = "trace", skip_all)]
958
36
    pub fn terminate(&self) {
959
36
        let _ = self.send_control(CtrlMsg::Shutdown);
960
36
    }
961

            
962
    /// Tell the reactor that the circuit with the given ID has gone away.
963
    #[instrument(level = "trace", skip_all)]
964
430
    pub fn close_circuit(&self, circid: CircId) -> Result<()> {
965
430
        self.send_control(CtrlMsg::CloseCircuit(circid))?;
966
322
        Ok(())
967
430
    }
968

            
969
    /// Return a future that will resolve once this channel has closed.
970
    ///
971
    /// Note that this method does not _cause_ the channel to shut down on its own.
972
64
    pub fn wait_for_close(
973
64
        &self,
974
64
    ) -> impl Future<Output = StdResult<CloseInfo, ClosedUnexpectedly>> + Send + Sync + 'static + use<>
975
    {
976
64
        self.reactor_closed_rx
977
64
            .clone()
978
64
            .into_future()
979
90
            .map(|recv| match recv {
980
32
                Ok(Ok(info)) => Ok(info),
981
20
                Ok(Err(e)) => Err(ClosedUnexpectedly::ReactorError(e)),
982
12
                Err(oneshot_broadcast::SenderDropped) => Err(ClosedUnexpectedly::ReactorDropped),
983
64
            })
984
64
    }
985

            
986
    /// Install a [`CircuitPadder`](client::CircuitPadder) for this channel.
987
    ///
988
    /// Replaces any previous padder installed.
989
    #[cfg(feature = "circ-padding-manual")]
990
    pub async fn start_padding(self: &Arc<Self>, padder: client::CircuitPadder) -> Result<()> {
991
        self.set_padder_impl(Some(padder)).await
992
    }
993

            
994
    /// Remove any [`CircuitPadder`](client::CircuitPadder) installed for this channel.
995
    ///
996
    /// Does nothing if there was not a padder installed there.
997
    #[cfg(feature = "circ-padding-manual")]
998
    pub async fn stop_padding(self: &Arc<Self>) -> Result<()> {
999
        self.set_padder_impl(None).await
    }
    /// Replace the [`CircuitPadder`](client::CircuitPadder) installed for this channel with `padder`.
    #[cfg(feature = "circ-padding-manual")]
    async fn set_padder_impl(
        self: &Arc<Self>,
        padder: Option<client::CircuitPadder>,
    ) -> Result<()> {
        let (tx, rx) = oneshot::channel();
        let msg = CtrlMsg::SetChannelPadder { padder, sender: tx };
        self.control
            .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(Arc::new(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.
82
fn check_id_match_helper<T, U>(my_ident: &T, wanted_ident: &U) -> Result<()>
82
where
82
    T: HasRelayIds + ?Sized,
82
    U: HasRelayIds + ?Sized,
{
120
    for desired in wanted_ident.identities() {
120
        let id_type = desired.id_type();
120
        match my_ident.identity(id_type) {
120
            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
                )));
            }
        }
    }
74
    Ok(())
82
}
impl HasRelayIds for Channel {
4947
    fn identity(
4947
        &self,
4947
        key_type: tor_linkspec::RelayIdType,
4947
    ) -> Option<tor_linkspec::RelayIdRef<'_>> {
4947
        self.peer_id.identity(key_type)
4947
    }
}
/// 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.
53
    pub(crate) fn check_agrees_with_type(
53
        &self,
53
        channel_type: ChannelType,
53
    ) -> StdResult<(), tor_error::Bug> {
        use ChannelType::*;
        use circmap::CircIdRange::*;
53
        match (channel_type, self) {
53
            (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`")),
        }
53
        Ok(())
53
    }
}
/// Make some fake channel details (for testing only!)
#[cfg(any(test, feature = "testing"))]
1246
fn fake_channel_details() -> Arc<ChannelDetails> {
1246
    let unused_since = AtomicOptTimestamp::new();
1246
    Arc::new(ChannelDetails {
1246
        unused_since,
1246
        memquota: crate::util::fake_mq(),
1246
    })
1246
}
/// 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
1244
pub(crate) fn fake_mpsc() -> (CellTx, CellRx) {
1244
    let (tx, rx) = crate::fake_mpsc(CHANNEL_BUFFER_SIZE);
    #[cfg(feature = "circ-padding")]
1244
    let (tx, rx) = counting_streams::channel(tx, rx);
1244
    (tx, rx)
1244
}
#[cfg(test)]
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::*;
    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.
    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(Arc::new(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());
        });
    }
}