//! Code for talking directly (over a TLS connection) to a Tor client or relay.

//!

//! Channels form the basis of the rest of the Tor protocol: they are

//! the only way for two Tor instances to talk.

//!

//! Channels are not useful directly for application requests: after

//! making a channel, it needs to get used to build circuits, and the

//! circuits are used to anonymize streams.  The streams are the

//! objects corresponding to directory requests.

//!

//! In general, you shouldn't try to manage channels on your own;

//! use the `tor-chanmgr` crate instead.

//!

//! To launch a channel:

//!

//!  * Create a TLS connection as an object that implements AsyncRead +

//!    AsyncWrite + StreamOps, and pass it to a channel builder. This will

//!    yield an [crate::client::channel::handshake::ClientInitiatorHandshake] that represents

//!    the state of the handshake.

//!  * Call [crate::client::channel::handshake::ClientInitiatorHandshake::connect] on the result

//!    to negotiate the rest of the handshake.  This will verify

//!    syntactic correctness of the handshake, but not its cryptographic

//!    integrity.

//!  * Call handshake::UnverifiedChannel::check on the result.  This

//!    finishes the cryptographic checks.

//!  * Call handshake::VerifiedChannel::finish on the result. This

//!    completes the handshake and produces an open channel and Reactor.

//!  * Launch an asynchronous task to call the reactor's run() method.

//!

//! One you have a running channel, you can create circuits on it with

//! its [Channel::new_tunnel] method.  See

//! [crate::client::circuit::PendingClientTunnel] for information on how to

//! proceed from there.

//!

//! # Design

//!

//! For now, this code splits the channel into two pieces: a "Channel"

//! object that can be used by circuits to write cells onto the

//! channel, and a "Reactor" object that runs as a task in the

//! background, to read channel cells and pass them to circuits as

//! appropriate.

//!

//! I'm not at all sure that's the best way to do that, but it's what

//! I could think of.

//!

//! # Limitations

//!

//! TODO: There is no rate limiting or fairness.

/// The size of the channel buffer for communication between `Channel` and its reactor.

pub const CHANNEL_BUFFER_SIZE: usize = 128;

mod circmap;

mod handler;

pub(crate) mod handshake;

pub mod kist;

mod msg;

pub mod padding;

pub mod params;

mod reactor;

mod unique_id;

pub use crate::channel::params::*;

pub(crate) use crate::channel::reactor::Reactor;

use crate::channel::reactor::{BoxedChannelSink, BoxedChannelStream};

pub use crate::channel::unique_id::UniqId;

use crate::client::circuit::PendingClientTunnel;

use crate::client::circuit::padding::{PaddingController, QueuedCellPaddingInfo};

use crate::memquota::{ChannelAccount, CircuitAccount, SpecificAccount as _};

use crate::peer::PeerInfo;

use crate::util::err::ChannelClosed;

use crate::util::oneshot_broadcast;

use crate::util::timeout::TimeoutEstimator;

use crate::util::ts::AtomicOptTimestamp;

use crate::{ClockSkew, client};

use crate::{Error, Result};

use cfg_if::cfg_if;

use reactor::BoxedChannelStreamOps;

use safelog::{MaybeSensitive, sensitive as sv};

use std::future::{Future, IntoFuture};

use std::net::IpAddr;

use std::pin::Pin;

use std::sync::{Mutex, MutexGuard};

use std::time::Duration;

use tor_cell::chancell::ChanMsg;

use tor_cell::chancell::{AnyChanCell, CircId, msg::Netinfo, msg::PaddingNegotiate};

use tor_error::internal;

use tor_linkspec::{HasRelayIds, OwnedChanTarget};

use tor_memquota::mq_queue::{self, ChannelSpec as _, MpscSpec};

use tor_rtcompat::{CoarseTimeProvider, DynTimeProvider, SleepProvider};

#[cfg(feature = "circ-padding")]

use tor_async_utils::counting_streams::{self, CountingSink, CountingStream};

#[cfg(feature = "relay")]

use crate::circuit::CircuitRxReceiver;

/// Imports that are re-exported pub if feature `testing` is enabled

///

/// Putting them together in a little module like this allows us to select the

/// visibility for all of these things together.

mod testing_exports {

    #![allow(unreachable_pub)]

    pub use super::reactor::CtrlMsg;

    pub use crate::circuit::celltypes::CreateResponse;

}

#[cfg(feature = "testing")]

pub use testing_exports::*;

#[cfg(not(feature = "testing"))]

use testing_exports::*;

use asynchronous_codec;

use futures::channel::mpsc;

use futures::io::{AsyncRead, AsyncWrite};

use oneshot_fused_workaround as oneshot;

use educe::Educe;

use futures::{FutureExt as _, Sink};

use std::result::Result as StdResult;

use std::sync::Arc;

use std::task::{Context, Poll};

use tracing::{instrument, trace};

// reexport

pub use super::client::channel::handshake::ClientInitiatorHandshake;

#[cfg(feature = "relay")]

pub use super::relay::channel::handshake::RelayInitiatorHandshake;

use crate::channel::unique_id::CircUniqIdContext;

use kist::KistParams;

/// This indicate what type of channel it is. It allows us to decide for the correct channel cell

/// state machines and authentication process (if any).

///

/// It is created when a channel is requested for creation which means the subsystem wanting to

/// open a channel needs to know what type it wants.

#[derive(Clone, Copy, Debug, derive_more::Display)]

#[non_exhaustive]

pub enum ChannelType {

    /// Client: Initiated from a client to a relay. Client is unauthenticated and relay is

    /// authenticated.

    ClientInitiator,

    /// Relay: Initiating as a relay to a relay. Both sides are authenticated.

    RelayInitiator,

    /// Relay: Responding as a relay to a relay or client. Authenticated or Unauthenticated.

    RelayResponder {

        /// Indicate if the channel is authenticated. Responding as a relay can be either from a

        /// Relay (authenticated) or a Client/Bridge (Unauthenticated). We only know this

        /// information once the handshake is completed.

        ///

        /// This side is always authenticated, the other side can be if a relay or not if

        /// bridge/client. This is set to false unless we end up authenticating the other side

        /// meaning a relay.

        authenticated: bool,

    },

}

impl ChannelType {

    /// Set that this channel type is now authenticated. This only applies to RelayResponder.

}

/// A channel cell frame used for sending and receiving cells on a channel. The handler takes care

/// of the cell codec transition depending in which state the channel is.

///

/// ChannelFrame is used to basically handle all in and outbound cells on a channel for its entire

/// lifetime.

pub(crate) type ChannelFrame<T> = asynchronous_codec::Framed<T, handler::ChannelCellHandler>;

/// An entry in a channel's queue of cells to be flushed.

pub(crate) type ChanCellQueueEntry = (AnyChanCell, Option<QueuedCellPaddingInfo>);

/// Helper: Return a new channel frame [ChannelFrame] from an object implementing AsyncRead + AsyncWrite. In the

/// tor context, it is always a TLS stream.

///

/// The ty (type) argument needs to be able to transform into a [handler::ChannelCellHandler] which would

/// generally be a [ChannelType].

{

/// Canonical state between this channel and its peer. This is inferred from the [`Netinfo`]

/// received during the channel handshake.

///

/// A connection is "canonical" if the TCP connection's peer IP address matches an address

/// that the relay itself claims in its [`Netinfo`] cell.

#[derive(Debug)]

pub(crate) struct Canonicity {

    /// The peer has proven this connection is canonical for its address: at least one NETINFO "my

    /// address" matches the observed TCP peer address.

    pub(crate) peer_is_canonical: bool,

    /// We appear canonical from the peer's perspective: its NETINFO "other address" matches our

    /// advertised relay address.

    pub(crate) canonical_to_peer: bool,

}

impl Canonicity {

    /// Using a [`Netinfo`], build the canonicity object with the given addresses.

    ///

    /// The `my_addrs` are the advertised address of this relay or empty if a client/bridge as they

    /// do not advertise or expose a reachable address.

    ///

    /// The `peer_addr` is the IP address we believe the peer has. In other words, it is either the

    /// IP we used to connect to or the address we see in the accept() phase of the connection.

    ///

    /// It can be None if we used a non-IP address to connect to the peer (PT).

        Self {

            // The "other addr" (our address as seen by the peer) matches the one we advertised.

            // The "my addresses" (the peer addresses that it claims to have) matches the one we

            // see on the connection or that we attempted to connect to.

        }

    /// Construct a fully canonical object.

    #[cfg(any(test, feature = "testing"))]

}

/// An open client channel, ready to send and receive Tor cells.

///

/// A channel is a direct connection to a Tor relay, implemented using TLS.

///

/// This struct is a frontend that can be used to send cells

/// and otherwise control the channel.  The main state is

/// in the Reactor object.

///

/// (Users need a mutable reference because of the types in `Sink`, and

/// ultimately because `cell_tx: mpsc::Sender` doesn't work without mut.

///

/// # Channel life cycle

///

/// Channels can be created directly here through a channel builder (client or relay) API.

/// For a higher-level API (with better support for TLS, pluggable transports,

/// and channel reuse) see the `tor-chanmgr` crate.

///

/// After a channel is created, it will persist until it is closed in one of

/// four ways:

///    1. A remote error occurs.

///    2. The other side of the channel closes the channel.

///    3. Someone calls [`Channel::terminate`] on the channel.

///    4. The last reference to the `Channel` is dropped. (Note that every circuit

///       on a `Channel` keeps a reference to it, which will in turn keep the

///       channel from closing until all those circuits have gone away.)

///

/// Note that in cases 1-3, the [`Channel`] object itself will still exist: it

/// will just be unusable for most purposes.  Most operations on it will fail

/// with an error.

pub struct Channel {

    /// The channel type.

    #[expect(unused)] // TODO: Remove once used.

    channel_type: ChannelType,

    /// A channel used to send control messages to the Reactor.

    control: mpsc::UnboundedSender<CtrlMsg>,

    /// A channel used to send cells to the Reactor.

    cell_tx: CellTx,

    /// A receiver that indicates whether the channel is closed.

    ///

    /// Awaiting will return a `CancelledError` event when the reactor is dropped.

    /// Read to decide if operations may succeed, and is returned by `wait_for_close`.

    reactor_closed_rx: oneshot_broadcast::Receiver<Result<CloseInfo>>,

    /// Padding controller, used to report when data is queued for this channel.

    padding_ctrl: PaddingController,

    /// A unique identifier for this channel.

    unique_id: UniqId,

    /// Target identity and address information for this peer.

    peer_id: OwnedChanTarget,

    /// Validated information for this peer.

    #[expect(unused)] // TODO(relay) Remove once used un choose_channel()

    peer: MaybeSensitive<PeerInfo>,

    /// The declared clock skew on this channel, at the time when this channel was

    /// created.

    clock_skew: ClockSkew,

    /// The time when this channel was successfully completed

    opened_at: coarsetime::Instant,

    /// Mutable state used by the `Channel.

    mutable: Mutex<MutableDetails>,

    /// Information shared with the reactor

    details: Arc<ChannelDetails>,

    /// Canonicity of this channel.

    canonicity: Canonicity,

}

/// This is information shared between the reactor and the frontend (`Channel` object).

///

/// `control` can't be here because we rely on it getting dropped when the last user goes away.

#[derive(Debug)]

pub(crate) struct ChannelDetails {

    /// Since when the channel became unused.

    ///

    /// If calling `time_since_update` returns None,

    /// this channel is still in use by at least one circuit.

    ///

    /// Set by reactor when a circuit is added or removed.

    /// Read from `Channel::duration_unused`.

    unused_since: AtomicOptTimestamp,

    /// Memory quota account

    ///

    /// This is here partly because we need to ensure it lives as long as the channel,

    /// as otherwise the memquota system will tear the account down.

    #[allow(dead_code)]

    memquota: ChannelAccount,

}

/// Mutable details (state) used by the `Channel` (frontend)

#[derive(Debug, Default)]

struct MutableDetails {

    /// State used to control padding

    padding: PaddingControlState,

}

/// State used to control padding

///

/// We store this here because:

///

///  1. It must be per-channel, because it depends on channel usage.  So it can't be in

///     (for example) `ChannelPaddingInstructionsUpdate`.

///

///  2. It could be in the channel manager's per-channel state but (for code flow reasons

///     there, really) at the point at which the channel manager concludes for a pending

///     channel that it ought to update the usage, it has relinquished the lock on its own data

///     structure.

///     And there is actually no need for this to be global: a per-channel lock is better than

///     reacquiring the global one.

///

///  3. It doesn't want to be in the channel reactor since that's super hot.

///

/// See also the overview at [`tor_proto::channel::padding`](padding)

#[derive(Debug, Educe)]

#[educe(Default)]

enum PaddingControlState {

    /// No usage of this channel, so far, implies sending or negotiating channel padding.

    ///

    /// This means we do not send (have not sent) any `ChannelPaddingInstructionsUpdates` to the reactor,

    /// with the following consequences:

    ///

    ///  * We don't enable our own padding.

    ///  * We don't do any work to change the timeout distribution in the padding timer,

    ///    (which is fine since this timer is not enabled).

    ///  * We don't send any PADDING_NEGOTIATE cells.  The peer is supposed to come to the

    ///    same conclusions as us, based on channel usage: it should also not send padding.

    #[educe(Default)]

    UsageDoesNotImplyPadding {

        /// The last padding parameters (from reparameterize)

        ///

        /// We keep this so that we can send it if and when

        /// this channel starts to be used in a way that implies (possibly) sending padding.

        padding_params: ChannelPaddingInstructionsUpdates,

    },

    /// Some usage of this channel implies possibly sending channel padding

    ///

    /// The required padding timer, negotiation cell, etc.,

    /// have been communicated to the reactor via a `CtrlMsg::ConfigUpdate`.

    ///

    /// Once we have set this variant, it remains this way forever for this channel,

    /// (the spec speaks of channels "only used for" certain purposes not getting padding).

    PaddingConfigured,

}

use PaddingControlState as PCS;

cfg_if! {

    if #[cfg(feature="circ-padding")] {

        /// Implementation type for a ChannelSender.

        type CellTx = CountingSink<mq_queue::Sender<ChanCellQueueEntry, mq_queue::MpscSpec>>;

        /// Implementation type for a cell queue held by a reactor.

        type CellRx = CountingStream<mq_queue::Receiver<ChanCellQueueEntry, mq_queue::MpscSpec>>;

    } else {

        /// Implementation type for a ChannelSender.

        type CellTx = mq_queue::Sender<ChanCellQueueEntry, mq_queue::MpscSpec>;

        /// Implementation type for a cell queue held by a reactor.

        type CellRx = mq_queue::Receiver<ChanCellQueueEntry, mq_queue::MpscSpec>;

    }

}

/// A handle to a [`Channel`]` that can be used, by circuits, to send channel cells.

#[derive(Debug)]

pub(crate) struct ChannelSender {

    /// MPSC sender to send cells.

    cell_tx: CellTx,

    /// A receiver used to check if the channel is closed.

    reactor_closed_rx: oneshot_broadcast::Receiver<Result<CloseInfo>>,

    /// Unique ID for this channel. For logging.

    unique_id: UniqId,

    /// Padding controller for this channel:

    /// used to report when we queue data that will eventually wind up on the channel.

    padding_ctrl: PaddingController,

}

impl ChannelSender {

    /// Check whether a cell type is permissible to be _sent_ on an

    /// open client channel.

        use tor_cell::chancell::msg::AnyChanMsg::*;

            Certs(_) | Versions(_) | Authenticate(_) | AuthChallenge(_) | Netinfo(_) => {

            }

        }

    /// Obtain a reference to the `ChannelSender`'s [`DynTimeProvider`]

    ///

    /// (This can sometimes be used to avoid having to keep

    /// a separate clone of the time provider.)

        cfg_if! {

            if #[cfg(feature="circ-padding")] {

            } else {

                self.cell_tx.time_provider()

            }

        }

    /// Return an approximate count of the number of outbound cells queued for this channel.

    ///

    /// This count is necessarily approximate,

    /// because the underlying count can be modified by other senders and receivers

    /// between when this method is called and when its return value is used.

    ///

    /// Does not include cells that have already been passed to the TLS connection.

    ///

    /// Circuit padding uses this count to determine

    /// when messages are already outbound for the first hop of a circuit.

    #[cfg(feature = "circ-padding")]

    /// Note that a cell has been queued that will eventually be placed onto this sender.

    ///

    /// We use this as an input for padding machines.

}

impl Sink<ChanCellQueueEntry> for ChannelSender {

    type Error = Error;

        {

            use tor_cell::chancell::msg::AnyChanMsg::*;

                    channel_id = %this.unique_id,

                ),

            }

        }

}

impl Channel {

    /// Construct a channel and reactor.

    ///

    /// Internal method, called to finalize the channel when we've

    /// sent our netinfo cell, received the peer's netinfo cell, and

    /// we're finally ready to create circuits.

    ///

    /// Quick note on the allow clippy. This is has one call site so for now, it is fine that we

    /// bust the mighty 7 arguments.

    #[allow(clippy::too_many_arguments)] // TODO consider if we want a builder

    {

        use circmap::{CircIdRange, CircMap};

        };

        #[cfg(feature = "circ-padding")]

        // We might be using experimental maybenot padding; this creates the padding framework for that.

        //

        // TODO: This backend is currently optimized for circuit padding,

        // so it might allocate a bit more than necessary to account for multiple hops.

        // We should tune it when we deploy padding in production.

        // We start disabled; the channel manager will `reconfigure` us soon after creation.

        cfg_if! {

            if #[cfg(feature = "circ-padding")] {

                use crate::util::sink_blocker::{SinkBlocker,CountingPolicy};

            }

        }

    /// Return a process-unique identifier for this channel.

    /// Return a reference to the memory tracking account for this Channel

    /// Obtain a reference to the `Channel`'s [`DynTimeProvider`]

    ///

    /// (This can sometimes be used to avoid having to keep

    /// a separate clone of the time provider.)

        cfg_if! {

            if #[cfg(feature="circ-padding")] {

            } else {

                self.cell_tx.time_provider()

            }

        }

    /// Return an OwnedChanTarget representing the actual handshake used to

    /// create this channel.

    /// Return the amount of time that has passed since this channel became open.

    /// Return a ClockSkew declaring how much clock skew the other side of this channel

    /// claimed that we had when we negotiated the connection.

    /// Send a control message

    #[instrument(level = "trace", skip_all)]

    /// Acquire the lock on `mutable` (and handle any poison error)

    /// Specify that this channel should do activities related to channel padding

    ///

    /// Initially, the channel does nothing related to channel padding:

    /// it neither sends any padding, nor sends any PADDING_NEGOTIATE cells.

    ///

    /// After this function has been called, it will do both,

    /// according to the parameters specified through `reparameterize`.

    /// Note that this might include *disabling* padding

    /// (for example, by sending a `PADDING_NEGOTIATE`).

    ///

    /// Idempotent.

    ///

    /// There is no way to undo the effect of this call.

    #[instrument(level = "trace", skip_all)]

            PCS::UsageDoesNotImplyPadding {

            } => {

                // Well, apparently the channel usage *does* imply padding now,

                // so we need to (belatedly) enable the timer,

                // send the padding negotiation cell, etc.

                // Except, maybe the padding we would be requesting is precisely default,

                // so we wouldn't actually want to send that cell.

                }

            }

        }

    /// Reparameterise (update parameters; reconfigure)

    ///

    /// Returns `Err` if the channel was closed earlier

    #[instrument(level = "trace", skip_all)]

            PCS::PaddingConfigured => {

            }

        }

    /// Update the KIST parameters.

    ///

    /// Returns `Err` if the channel is closed.

    #[instrument(level = "trace", skip_all)]

    /// Return an error if this channel is somehow mismatched with the

    /// given target.

    /// Return true if this channel is closed and therefore unusable.

    /// Return true iff this channel is considered canonical by us.

    /// Return true if we think the peer considers this channel as canonical.

    /// If the channel is not in use, return the amount of time

    /// it has had with no circuits.

    ///

    /// Return `None` if the channel is currently in use.

    /// Return a new [`ChannelSender`] to transmit cells on this channel.

    /// Return a newly allocated PendingClientTunnel object with

    /// a corresponding tunnel reactor. A circuit ID is allocated, but no

    /// messages are sent, and no cryptography is done.

    ///

    /// To use the results of this method, call Reactor::run() in a

    /// new task, then use the methods of

    /// [crate::client::circuit::PendingClientTunnel] to build the circuit.

    #[instrument(level = "trace", skip_all)]

        if self.is_closing() {

            return Err(ChannelClosed.into());

        }

        let time_prov = self.time_provider().clone();

        let memquota = CircuitAccount::new(&self.details.memquota)?;

        // TODO: blocking is risky, but so is unbounded.

        let (sender, receiver) =

            MpscSpec::new(128).new_mq(time_prov.clone(), memquota.as_raw_account())?;

        let (createdsender, createdreceiver) = oneshot::channel::<CreateResponse>();

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

        self.send_control(CtrlMsg::AllocateCircuit {

            created_sender: createdsender,

            sender,

            tx,

        })?;

        let (id, circ_unique_id, padding_ctrl, padding_stream) =

            rx.await.map_err(|_| ChannelClosed)??;

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

        Ok(PendingClientTunnel::new(

            id,

            self.clone(),

            createdreceiver,

            receiver,

            circ_unique_id,

            time_prov,

            memquota,

            padding_ctrl,

            padding_stream,

            timeouts,

        ))

    /// Return a newly allocated outbound relay circuit with.

    ///

    /// A circuit ID is allocated, but no messages are sent, and no cryptography is done.

    ///

    // TODO(relay): this duplicates much of new_tunnel above, but I expect

    // the implementations to diverge once we introduce a new CtrlMsg for

    // allocating relay circuits.

    #[cfg(feature = "relay")]

        // TODO: blocking is risky, but so is unbounded.

        // TODO(relay): I don't think we need circuit-level padding on this side of the circuit.

        // This just drops the padding controller and corresponding event stream,

        // but maybe it would be better to just not set it up in the first place?

        // This suggests we might need a different control command for allocating

        // the outbound relay circuits...

    /// Shut down this channel immediately, along with all circuits that

    /// are using it.

    ///

    /// Note that other references to this channel may exist.  If they

    /// do, they will stop working after you call this function.

    ///

    /// It's not necessary to call this method if you're just done

    /// with a channel: the channel should close on its own once nothing

    /// is using it any more.

    #[instrument(level = "trace", skip_all)]

    /// Tell the reactor that the circuit with the given ID has gone away.

    #[instrument(level = "trace", skip_all)]

    /// Return a future that will resolve once this channel has closed.

    ///

    /// Note that this method does not _cause_ the channel to shut down on its own.

    {

    /// Install a [`CircuitPadder`](client::CircuitPadder) for this channel.

    ///

    /// Replaces any previous padder installed.

    #[cfg(feature = "circ-padding-manual")]

    /// Remove any [`CircuitPadder`](client::CircuitPadder) installed for this channel.

    ///

    /// Does nothing if there was not a padder installed there.

    #[cfg(feature = "circ-padding-manual")]

    /// Replace the [`CircuitPadder`](client::CircuitPadder) installed for this channel with `padder`.

    #[cfg(feature = "circ-padding-manual")]

    /// 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")]

        // This will make rx trigger immediately.

}

/// 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.

{

            }

            None => {

            }

        }

    }

impl HasRelayIds for Channel {

}

/// 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),

}

/// Make some fake channel details (for testing only!)

#[cfg(any(test, feature = "testing"))]

/// 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

    #[cfg(feature = "circ-padding")]

#[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 {

            channel_type,

            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());

        });

    }

}