//! Multi-hop paths over the Tor network.

//!

//! Right now, we only implement "client circuits" -- also sometimes

//! called "origin circuits".  A client circuit is one that is

//! constructed by this Tor instance, and used in its own behalf to

//! send data over the Tor network.

//!

//! Each circuit has multiple hops over the Tor network: each hop

//! knows only the hop before and the hop after.  The client shares a

//! separate set of keys with each hop.

//!

//! To build a circuit, first create a [crate::channel::Channel], then

//! call its [crate::channel::Channel::new_tunnel] method.  This yields

//! a [PendingClientTunnel] object that won't become live until you call

//! one of the methods

//! (typically [`PendingClientTunnel::create_firsthop`])

//! that extends it to its first hop.  After you've

//! done that, you can call [`ClientCirc::extend`] on the tunnel to

//! build it into a multi-hop tunnel.  Finally, you can use

//! [ClientTunnel::begin_stream] to get a Stream object that can be used

//! for anonymized data.

//!

//! # Implementation

//!

//! Each open circuit has a corresponding Reactor object that runs in

//! an asynchronous task, and manages incoming cells from the

//! circuit's upstream channel.  These cells are either RELAY cells or

//! DESTROY cells.  DESTROY cells are handled immediately.

//! RELAY cells are either for a particular stream, in which case they

//! get forwarded to a RawCellStream object, or for no particular stream,

//! in which case they are considered "meta" cells (like EXTENDED2)

//! that should only get accepted if something is waiting for them.

//!

//! # Limitations

//!

//! This is client-only.

pub(crate) mod halfcirc;

#[cfg(feature = "hs-common")]

pub mod handshake;

#[cfg(not(feature = "hs-common"))]

pub(crate) mod handshake;

pub(crate) mod padding;

pub(super) mod path;

use crate::channel::Channel;

use crate::circuit::circhop::{HopNegotiationType, HopSettings};

use crate::circuit::{CircuitRxReceiver, celltypes::*};

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

use crate::client::CircuitPadder;

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

use crate::client::reactor::{CircuitHandshake, CtrlCmd, CtrlMsg, Reactor};

use crate::crypto::cell::HopNum;

use crate::crypto::handshake::ntor_v3::NtorV3PublicKey;

use crate::memquota::CircuitAccount;

use crate::util::skew::ClockSkew;

use crate::{Error, Result};

use derive_deftly::Deftly;

use educe::Educe;

use path::HopDetail;

use tor_cell::chancell::{

    CircId,

    msg::{self as chanmsg},

};

use tor_error::{bad_api_usage, internal, into_internal};

use tor_linkspec::{CircTarget, LinkSpecType, OwnedChanTarget, RelayIdType};

use tor_protover::named;

use tor_rtcompat::DynTimeProvider;

use crate::circuit::UniqId;

use super::{ClientTunnel, TargetHop};

use futures::channel::mpsc;

use oneshot_fused_workaround as oneshot;

use futures::FutureExt as _;

use std::collections::HashMap;

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

use tor_memquota::derive_deftly_template_HasMemoryCost;

use crate::crypto::handshake::ntor::NtorPublicKey;

#[cfg(test)]

use crate::stream::{StreamMpscReceiver, StreamMpscSender};

pub use crate::crypto::binding::CircuitBinding;

pub use path::{Path, PathEntry};

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

pub const CIRCUIT_BUFFER_SIZE: usize = 128;

// TODO: export this from the top-level instead (it's not client-specific).

pub use crate::circuit::CircParameters;

// TODO(relay): reexport this from somewhere else (it's not client-specific)

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

/// A subclass of ChanMsg that can correctly arrive on a live client

/// circuit (one where a CREATED* has been received).

#[derive(Debug, Deftly)]

#[allow(unreachable_pub)] // Only `pub` with feature `testing`; otherwise, visible in crate

#[derive_deftly(HasMemoryCost)]

#[derive_deftly(RestrictedChanMsgSet)]

#[deftly(usage = "on an open client circuit")]

pub(super) enum ClientCircChanMsg {

    /// A relay cell telling us some kind of remote command from some

    /// party on the circuit.

    Relay(chanmsg::Relay),

    /// A cell telling us to destroy the circuit.

    Destroy(chanmsg::Destroy),

    // Note: RelayEarly is not valid for clients!

}

#[derive(Debug)]

/// A circuit that we have constructed over the Tor network.

///

/// # Circuit life cycle

///

/// `ClientCirc`s are created in an initially unusable state using [`Channel::new_tunnel`],

/// which returns a [`PendingClientTunnel`].  To get a real (one-hop) tunnel from

/// one of these, you invoke one of its `create_firsthop` methods (typically

/// [`create_firsthop_fast()`](PendingClientTunnel::create_firsthop_fast) or

/// [`create_firsthop()`](PendingClientTunnel::create_firsthop)).

/// Then, to add more hops to the circuit, you can call

/// [`extend()`](ClientCirc::extend) on it.

///

/// For higher-level APIs, see the `tor-circmgr` crate: the ones here in

/// `tor-proto` are probably not what you need.

///

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

/// five ways:

///    1. A remote error occurs.

///    2. Some hop on the circuit sends a `DESTROY` message to tear down the

///       circuit.

///    3. The circuit's channel is closed.

///    4. Someone calls [`ClientTunnel::terminate`] on the tunnel owning the circuit.

///    5. The last reference to the `ClientCirc` is dropped. (Note that every stream

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

///       circuit from closing until all those streams have gone away.)

///

/// Note that in cases 1-4 the [`ClientCirc`] object itself will still exist: it

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

/// with an error.

//

// Effectively, this struct contains two Arcs: one for `path` and one for

// `control` (which surely has something Arc-like in it).  We cannot unify

// these by putting a single Arc around the whole struct, and passing

// an Arc strong reference to the `Reactor`, because then `control` would

// not be dropped when the last user of the circuit goes away.  We could

// make the reactor have a weak reference but weak references are more

// expensive to dereference.

//

// Because of the above, cloning this struct is always going to involve

// two atomic refcount changes/checks.  Wrapping it in another Arc would

// be overkill.

//

pub struct ClientCirc {

    /// Mutable state shared with the `Reactor`.

    pub(super) mutable: Arc<TunnelMutableState>,

    /// A unique identifier for this circuit.

    unique_id: UniqId,

    /// Channel to send control messages to the reactor.

    pub(super) control: mpsc::UnboundedSender<CtrlMsg>,

    /// Channel to send commands to the reactor.

    pub(super) command: mpsc::UnboundedSender<CtrlCmd>,

    /// A future that resolves to Cancelled once the reactor is shut down,

    /// meaning that the circuit is closed.

    #[cfg_attr(not(feature = "experimental-api"), allow(dead_code))]

    reactor_closed_rx: futures::future::Shared<oneshot::Receiver<void::Void>>,

    /// For testing purposes: the CircId, for use in peek_circid().

    #[cfg(test)]

    circid: CircId,

    /// Memory quota account

    pub(super) memquota: CircuitAccount,

    /// Time provider

    pub(super) time_provider: DynTimeProvider,

    /// Indicate if this reactor is a multi path or not. This is flagged at the very first

    /// LinkCircuit seen and never changed after.

    ///

    /// We can't just look at the number of legs because a multi path tunnel could have 1 leg only

    /// because the other(s) have collapsed.

    ///

    /// This is very important because it allows to make a quick efficient safety check by the

    /// circmgr higher level tunnel type without locking the mutable state or using the command

    /// channel.

    pub(super) is_multi_path: bool,

}

/// The mutable state of a tunnel, shared between [`ClientCirc`] and [`Reactor`].

///

/// NOTE(gabi): this mutex-inside-a-mutex might look suspicious,

/// but it is currently the best option we have for sharing

/// the circuit state with `ClientCirc` (and soon, with `ClientTunnel`).

/// In practice, these mutexes won't be accessed very often

/// (they're accessed for writing when a circuit is extended,

/// and for reading by the various `ClientCirc` APIs),

/// so they shouldn't really impact performance.

///

/// Alternatively, the circuit state information could be shared

/// outside the reactor through a channel (passed to the reactor via a `CtrlCmd`),

/// but in #1840 @opara notes that involves making the `ClientCirc` accessors

/// (`ClientCirc::path`, `ClientCirc::binding_key`, etc.)

/// asynchronous, which will significantly complicate their callsites,

/// which would in turn need to be made async too.

///

/// We should revisit this decision at some point, and decide whether an async API

/// would be preferable.

#[derive(Debug, Default)]

pub(super) struct TunnelMutableState(Mutex<HashMap<UniqId, Arc<MutableState>>>);

impl TunnelMutableState {

    /// Add the [`MutableState`] of a circuit.

        #[allow(unused)] // unused in non-debug builds

    /// Remove the [`MutableState`] of a circuit.

        #[allow(unused)] // unused in non-debug builds

    /// Return a [`Path`] object describing all the circuits in this tunnel.

    /// Return a list of [`Path`] objects describing the only circuit in this tunnel.

    ///

    /// Returns an error if the tunnel has more than one tunnel.

    //

    // TODO: replace Itertools::exactly_one() with a stdlib equivalent when there is one.

    //

    // See issue #48919 <https://github.com/rust-lang/rust/issues/48919>

    #[allow(unstable_name_collisions)]

        use itertools::Itertools as _;

    /// Return a description of the first hop of this circuit.

    ///

    /// Returns an error if a circuit with the specified [`UniqId`] doesn't exist.

    /// Returns `Ok(None)` if the specified circuit doesn't have any hops.

            #[cfg(feature = "hs-common")]

            path::HopDetail::Virtual => {

            }

    /// Return the [`HopNum`] of the last hop of the specified circuit.

    ///

    /// Returns an error if a circuit with the specified [`UniqId`] doesn't exist.

    ///

    /// See [`MutableState::last_hop_num`].

    /// Return the number of hops in the specified circuit.

    ///

    /// See [`MutableState::n_hops`].

}

/// The mutable state of a circuit.

#[derive(Educe, Default)]

#[educe(Debug)]

pub(super) struct MutableState(Mutex<CircuitState>);

impl MutableState {

    /// Add a hop to the path of this circuit.

    /// Get a copy of the circuit's current [`path::Path`].

    /// Return the cryptographic material used to prove knowledge of a shared

    /// secret with with `hop`.

        // NOTE: I'm not thrilled to have to copy this information, but we use

        // it very rarely, so it's not _that_ bad IMO.

    /// Return a description of the first hop of this circuit.

    /// Return the [`HopNum`] of the last hop of this circuit.

    ///

    /// NOTE: This function will return the [`HopNum`] of the hop

    /// that is _currently_ the last. If there is an extend operation in progress,

    /// the currently pending hop may or may not be counted, depending on whether

    /// the extend operation finishes before this call is done.

    /// Return the number of hops in this circuit.

    ///

    /// NOTE: This function will currently return only the number of hops

    /// _currently_ in the circuit. If there is an extend operation in progress,

    /// the currently pending hop may or may not be counted, depending on whether

    /// the extend operation finishes before this call is done.

}

/// The shared state of a circuit.

#[derive(Educe, Default)]

#[educe(Debug)]

pub(super) struct CircuitState {

    /// Information about this circuit's path.

    ///

    /// This is stored in an Arc so that we can cheaply give a copy of it to

    /// client code; when we need to add a hop (which is less frequent) we use

    /// [`Arc::make_mut()`].

    path: Arc<path::Path>,

    /// Circuit binding keys [q.v.][`CircuitBinding`] information for each hop

    /// in the circuit's path.

    ///

    /// NOTE: Right now, there is a `CircuitBinding` for every hop.  There's a

    /// fair chance that this will change in the future, and I don't want other

    /// code to assume that a `CircuitBinding` _must_ exist, so I'm making this

    /// an `Option`.

    #[educe(Debug(ignore))]

    binding: Vec<Option<CircuitBinding>>,

}

/// A ClientCirc that needs to send a create cell and receive a created* cell.

///

/// To use one of these, call `create_firsthop_fast()` or `create_firsthop()`

/// to negotiate the cryptographic handshake with the first hop.

pub struct PendingClientTunnel {

    /// A oneshot receiver on which we'll receive a CREATED* cell,

    /// or a DESTROY cell.

    recvcreated: oneshot::Receiver<CreateResponse>,

    /// The ClientCirc object that we can expose on success.

    circ: ClientCirc,

}

impl ClientCirc {

    /// Convert this `ClientCirc` into a single circuit [`ClientTunnel`].

    /// Return a description of the first hop of this circuit.

    ///

    /// # Panics

    ///

    /// Panics if there is no first hop.  (This should be impossible outside of

    /// the tor-proto crate, but within the crate it's possible to have a

    /// circuit with no hops.)

    /// Return a description of the last hop of the tunnel.

    ///

    /// Return None if the last hop is virtual.

    ///

    /// # Panics

    ///

    /// Panics if there is no last hop.  (This should be impossible outside of

    /// the tor-proto crate, but within the crate it's possible to have a

    /// circuit with no hops.)

    /// Return the [`HopNum`] of the last hop of this circuit.

    ///

    /// Returns an error if there is no last hop.  (This should be impossible outside of the

    /// tor-proto crate, but within the crate it's possible to have a circuit with no hops.)

    ///

    /// NOTE: This function will return the [`HopNum`] of the hop

    /// that is _currently_ the last. If there is an extend operation in progress,

    /// the currently pending hop may or may not be counted, depending on whether

    /// the extend operation finishes before this call is done.

    /// Return a [`TargetHop`] representing precisely the last hop of the circuit as in set as a

    /// HopLocation with its id and hop number.

    ///

    /// Return an error if there is no last hop.

    /// Return a list of [`Path`] objects describing all the circuits in this tunnel.

    ///

    /// Note that these `Path`s are not automatically updated if the underlying

    /// circuits are extended.

    /// Return a list of [`Path`] objects describing the only circuit in this tunnel.

    ///

    /// Returns an error if the tunnel has more than one tunnel.

    /// Return the time at which this circuit last had any open streams.

    ///

    /// Returns `None` if this circuit has never had any open streams,

    /// or if it currently has open streams.

    ///

    /// NOTE that the Instant returned by this method is not affected by

    /// any runtime mocking; it is the output of an ordinary call to

    /// `Instant::now()`.

    /// Get the clock skew claimed by the first hop of the circuit.

    ///

    /// See [`Channel::clock_skew()`].

    /// Return a reference to this circuit's memory quota account

    /// Return the cryptographic material used to prove knowledge of a shared

    /// secret with with `hop`.

    ///

    /// See [`CircuitBinding`] for more information on how this is used.

    ///

    /// Return None if we have no circuit binding information for the hop, or if

    /// the hop does not exist.

    #[cfg(feature = "hs-service")]

    /// Extend the circuit, via the most appropriate circuit extension handshake,

    /// to the chosen `target` hop.

        #![allow(deprecated)]

        // For now we use the simplest decision-making mechanism:

        // we use ntor_v3 whenever it is present; and otherwise we use ntor.

        //

        // This behavior is slightly different from C tor, which uses ntor v3

        // only whenever it want to send any extension in the circuit message.

        // But thanks to congestion control (named::FLOWCTRL_CC), we'll _always_

        // want to use an extension if we can, and so it doesn't make too much

        // sense to detect the case where we have no extensions.

        //

        // (As of April 2025, RELAY_NTORV3 is not yet listed as Required for relays

        // on the tor network, and so we cannot simply assume that everybody has it.)

        {

        } else {

        }

    /// Extend the circuit via the ntor handshake to a new target last

    /// hop.

    #[deprecated(since = "1.6.1", note = "Use extend instead.")]

        };

    /// Extend the circuit via the ntor handshake to a new target last

    /// hop.

    #[deprecated(since = "1.6.1", note = "Use extend instead.")]

        };

    /// Extend this circuit by a single, "virtual" hop.

    ///

    /// A virtual hop is one for which we do not add an actual network connection

    /// between separate hosts (such as Relays).  We only add a layer of

    /// cryptography.

    ///

    /// This is used to implement onion services: the client and the service

    /// both build a circuit to a single rendezvous point, and tell the

    /// rendezvous point to relay traffic between their two circuits.  Having

    /// completed a [`handshake`] out of band[^1], the parties each extend their

    /// circuits by a single "virtual" encryption hop that represents their

    /// shared cryptographic context.

    ///

    /// Once a circuit has been extended in this way, it is an error to try to

    /// extend it in any other way.

    ///

    /// [^1]: Technically, the handshake is only _mostly_ out of band: the

    ///     client sends their half of the handshake in an ` message, and the

    ///     service's response is inline in its `RENDEZVOUS2` message.

    //

    // TODO hs: let's try to enforce the "you can't extend a circuit again once

    // it has been extended this way" property.  We could do that with internal

    // state, or some kind of a type state pattern.

    #[cfg(feature = "hs-common")]

        use self::handshake::BoxedClientLayer;

        // TODO CGO: Possibly refactor this match into a separate method when we revisit this.

        };

    /// Install a [`CircuitPadder`] at the listed `hop`.

    ///

    /// Replaces any previous padder installed at that hop.

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

    /// Remove any [`CircuitPadder`] at the listed `hop`.

    ///

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

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

    /// Helper: replace the padder at `hop` with the provided `padder`, or with `None`.

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

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

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

    /// Return the number of hops in this circuit.

    ///

    /// NOTE: This function will currently return only the number of hops

    /// _currently_ in the circuit. If there is an extend operation in progress,

    /// the currently pending hop may or may not be counted, depending on whether

    /// the extend operation finishes before this call is done.

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

    ///

    /// Note that this method does not itself cause the circuit to shut down.

    ///

    /// TODO: Perhaps this should return some kind of status indication instead

    /// of just ()

}

impl PendingClientTunnel {

    /// Instantiate a new circuit object: used from Channel::new_tunnel().

    ///

    /// Does not send a CREATE* cell on its own.

    #[allow(clippy::too_many_arguments)]

    /// Extract the process-unique identifier for this pending circuit.

    /// Use the (questionable!) CREATE_FAST handshake to connect to the

    /// first hop of this circuit.

    ///

    /// There's no authentication in CRATE_FAST,

    /// so we don't need to know whom we're connecting to: we're just

    /// connecting to whichever relay the channel is for.

        // We know nothing about this relay, so we assume it supports no protocol capabilities at all.

        //

        // TODO: If we had a consensus, we could assume it supported all required-relay-protocols.

        // TODO prop364: When we implement CreateOneHop, we will want a Protocols argument here.

    /// Use the most appropriate handshake to connect to the first hop of this circuit.

    ///

    /// Note that the provided 'target' must match the channel's target,

    /// or the handshake will fail.

        #![allow(deprecated)]

        // (See note in ClientCirc::extend.)

        {

        } else {

        }

    /// Use the ntor handshake to connect to the first hop of this circuit.

    ///

    /// Note that the provided 'target' must match the channel's target,

    /// or the handshake will fail.

    #[deprecated(since = "1.6.1", note = "Use create_firsthop instead.")]

                handshake: CircuitHandshake::Ntor {

                    public_key: NtorPublicKey {

                    },

                },

            })

    /// Use the ntor_v3 handshake to connect to the first hop of this circuit.

    ///

    /// Assumes that the target supports ntor_v3. The caller should verify

    /// this before calling this function, e.g. by validating that the target

    /// has advertised ["Relay=4"](https://spec.torproject.org/tor-spec/subprotocol-versioning.html#relay).

    ///

    /// Note that the provided 'target' must match the channel's target,

    /// or the handshake will fail.

    #[deprecated(since = "1.6.1", note = "Use create_firsthop instead.")]

                handshake: CircuitHandshake::NtorV3 {

                    public_key: NtorV3PublicKey {

                    },

                },

            })

}

#[cfg(test)]

pub(crate) mod test {

    // @@ begin test lint list maintained by maint/add_warning @@

    #![allow(clippy::bool_assert_comparison)]

    #![allow(clippy::clone_on_copy)]

    #![allow(clippy::dbg_macro)]

    #![allow(clippy::mixed_attributes_style)]

    #![allow(clippy::print_stderr)]

    #![allow(clippy::print_stdout)]

    #![allow(clippy::single_char_pattern)]

    #![allow(clippy::unwrap_used)]

    #![allow(clippy::unchecked_time_subtraction)]

    #![allow(clippy::useless_vec)]

    #![allow(clippy::needless_pass_by_value)]

    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->

    use super::*;

    use crate::channel::test::{CodecResult, new_reactor};

    use crate::circuit::CircuitRxSender;

    use crate::client::circuit::padding::new_padding;

    use crate::client::stream::DataStream;

    #[cfg(feature = "hs-service")]

    use crate::client::stream::IncomingStreamRequestFilter;

    use crate::congestion::params::CongestionControlParams;

    use crate::congestion::test_utils::params::build_cc_vegas_params;

    use crate::crypto::cell::RelayCellBody;

    use crate::crypto::handshake::ntor_v3::NtorV3Server;

    use crate::memquota::SpecificAccount as _;

    use crate::stream::flow_ctrl::params::FlowCtrlParameters;

    use crate::util::DummyTimeoutEstimator;

    use assert_matches::assert_matches;

    use chanmsg::{AnyChanMsg, Created2, CreatedFast};

    use futures::channel::mpsc::{Receiver, Sender};

    use futures::io::{AsyncReadExt, AsyncWriteExt};

    use futures::sink::SinkExt;

    use futures::stream::StreamExt;

    use hex_literal::hex;

    use std::collections::{HashMap, VecDeque};

    use std::fmt::Debug;

    use std::time::Duration;

    use tor_basic_utils::test_rng::testing_rng;

    use tor_cell::chancell::{AnyChanCell, BoxedCellBody, ChanCell, ChanCmd, msg as chanmsg};

    use tor_cell::relaycell::extend::{self as extend_ext, CircRequestExt, CircResponseExt};

    use tor_cell::relaycell::msg::SendmeTag;

    use tor_cell::relaycell::{

        AnyRelayMsgOuter, RelayCellFormat, RelayCmd, StreamId, msg as relaymsg, msg::AnyRelayMsg,

    };

    use tor_cell::relaycell::{RelayMsg, UnparsedRelayMsg};

    use tor_linkspec::OwnedCircTarget;

    use tor_memquota::HasMemoryCost;

    use tor_rtcompat::Runtime;

    use tor_rtcompat::SpawnExt;

    use tracing::trace;

    use tracing_test::traced_test;

    #[cfg(feature = "conflux")]

    use {

        crate::client::reactor::ConfluxHandshakeResult,

        crate::util::err::ConfluxHandshakeError,

        futures::future::FusedFuture,

        futures::lock::Mutex as AsyncMutex,

        std::pin::Pin,

        std::result::Result as StdResult,

        tor_cell::relaycell::conflux::{V1DesiredUx, V1LinkPayload, V1Nonce},

        tor_cell::relaycell::msg::ConfluxLink,

        tor_rtmock::MockRuntime,

    };

    impl PendingClientTunnel {

        /// Testing only: Extract the circuit ID for this pending circuit.

        pub(crate) fn peek_circid(&self) -> CircId {

            self.circ.circid

        }

    }

    impl ClientCirc {

        /// Testing only: Extract the circuit ID of this circuit.

        pub(crate) fn peek_circid(&self) -> CircId {

            self.circid

        }

    }

    impl ClientTunnel {

        pub(crate) async fn resolve_last_hop(&self) -> TargetHop {

            let (sender, receiver) = oneshot::channel();

            let _ =

                self.as_single_circ()

                    .unwrap()

                    .command

                    .unbounded_send(CtrlCmd::ResolveTargetHop {

                        hop: TargetHop::LastHop,

                        done: sender,

                    });

            TargetHop::Hop(receiver.await.unwrap().unwrap())

        }

    }

    fn rmsg_to_ccmsg(id: Option<StreamId>, msg: relaymsg::AnyRelayMsg) -> AnyChanMsg {

        // TODO #1947: test other formats.

        let rfmt = RelayCellFormat::V0;

        let body: BoxedCellBody = AnyRelayMsgOuter::new(id, msg)

            .encode(rfmt, &mut testing_rng())

            .unwrap();

        let chanmsg = chanmsg::Relay::from(body);

        AnyChanMsg::Relay(chanmsg)

    }

    // Example relay IDs and keys

    const EXAMPLE_SK: [u8; 32] =

        hex!("7789d92a89711a7e2874c61ea495452cfd48627b3ca2ea9546aafa5bf7b55803");

    const EXAMPLE_PK: [u8; 32] =

        hex!("395cb26b83b3cd4b91dba9913e562ae87d21ecdd56843da7ca939a6a69001253");

    const EXAMPLE_ED_ID: [u8; 32] = [6; 32];

    const EXAMPLE_RSA_ID: [u8; 20] = [10; 20];

    /// Make an MPSC queue, of the type we use in Channels, but a fake one for testing

    #[cfg(test)]

    pub(crate) fn fake_mpsc<T: HasMemoryCost + Debug + Send>(

        buffer: usize,

    ) -> (StreamMpscSender<T>, StreamMpscReceiver<T>) {

        crate::fake_mpsc(buffer)

    }

    /// return an example OwnedCircTarget that can get used for an ntor handshake.

    fn example_target() -> OwnedCircTarget {

        let mut builder = OwnedCircTarget::builder();

        builder

            .chan_target()

            .ed_identity(EXAMPLE_ED_ID.into())

            .rsa_identity(EXAMPLE_RSA_ID.into());

        builder

            .ntor_onion_key(EXAMPLE_PK.into())

            .protocols("FlowCtrl=1-2".parse().unwrap())

            .build()

            .unwrap()

    }

    fn example_ntor_key() -> crate::crypto::handshake::ntor::NtorSecretKey {

        crate::crypto::handshake::ntor::NtorSecretKey::new(

            EXAMPLE_SK.into(),

            EXAMPLE_PK.into(),

            EXAMPLE_RSA_ID.into(),

        )

    }

    fn example_ntor_v3_key() -> crate::crypto::handshake::ntor_v3::NtorV3SecretKey {

        crate::crypto::handshake::ntor_v3::NtorV3SecretKey::new(

            EXAMPLE_SK.into(),

            EXAMPLE_PK.into(),

            EXAMPLE_ED_ID.into(),

        )

    }

    fn working_fake_channel<R: Runtime>(

        rt: &R,

    ) -> (Arc<Channel>, Receiver<AnyChanCell>, Sender<CodecResult>) {

        let (channel, chan_reactor, rx, tx) = new_reactor(rt.clone());

        rt.spawn(async {

            let _ignore = chan_reactor.run().await;

        })

        .unwrap();

        (channel, rx, tx)

    }

    /// Which handshake type to use.

    #[derive(Copy, Clone)]

    enum HandshakeType {

        Fast,

        Ntor,

        NtorV3,

    }

    #[allow(deprecated)]

    async fn test_create<R: Runtime>(rt: &R, handshake_type: HandshakeType, with_cc: bool) {

        // We want to try progressing from a pending circuit to a circuit

        // via a crate_fast handshake.

        use crate::crypto::handshake::{ServerHandshake, fast::CreateFastServer, ntor::NtorServer};

        let (chan, mut rx, _sink) = working_fake_channel(rt);

        let circid = CircId::new(128).unwrap();

        let (created_send, created_recv) = oneshot::channel();

        let (_circmsg_send, circmsg_recv) = fake_mpsc(64);

        let unique_id = UniqId::new(23, 17);

        let (padding_ctrl, padding_stream) = new_padding(DynTimeProvider::new(rt.clone()));

        let (pending, reactor) = PendingClientTunnel::new(

            circid,

            chan,

            created_recv,

            circmsg_recv,

            unique_id,

            DynTimeProvider::new(rt.clone()),

            CircuitAccount::new_noop(),

            padding_ctrl,

            padding_stream,

            Arc::new(DummyTimeoutEstimator),

        );

        rt.spawn(async {

            let _ignore = reactor.run().await;

        })

        .unwrap();

        // Future to pretend to be a relay on the other end of the circuit.

        let simulate_relay_fut = async move {

            let mut rng = testing_rng();

            let create_cell = rx.next().await.unwrap();

            assert_eq!(create_cell.circid(), Some(circid));

            let reply = match handshake_type {

                HandshakeType::Fast => {

                    let cf = match create_cell.msg() {

                        AnyChanMsg::CreateFast(cf) => cf,

                        other => panic!("{:?}", other),

                    };

                    let (_, rep) = CreateFastServer::server(

                        &mut rng,

                        &mut |_: &()| Some(()),

                        &[()],

                        cf.handshake(),

                    )

                    .unwrap();

                    CreateResponse::CreatedFast(CreatedFast::new(rep))

                }

                HandshakeType::Ntor => {

                    let c2 = match create_cell.msg() {

                        AnyChanMsg::Create2(c2) => c2,

                        other => panic!("{:?}", other),

                    };

                    let (_, rep) = NtorServer::server(

                        &mut rng,

                        &mut |_: &()| Some(()),

                        &[example_ntor_key()],

                        c2.body(),

                    )

                    .unwrap();

                    CreateResponse::Created2(Created2::new(rep))

                }

                HandshakeType::NtorV3 => {

                    let c2 = match create_cell.msg() {

                        AnyChanMsg::Create2(c2) => c2,

                        other => panic!("{:?}", other),

                    };

                    let mut reply_fn = if with_cc {

                        |client_exts: &[CircRequestExt]| {

                            let _ = client_exts

                                .iter()

                                .find(|e| matches!(e, CircRequestExt::CcRequest(_)))

                                .expect("Client failed to request CC");

                            // This needs to be aligned to test_utils params

                            // value due to validation that needs it in range.

                            Some(vec![CircResponseExt::CcResponse(

                                extend_ext::CcResponse::new(31),

                            )])

                        }

                    } else {

                        |_: &_| Some(vec![])

                    };

                    let (_, rep) = NtorV3Server::server(

                        &mut rng,

                        &mut reply_fn,

                        &[example_ntor_v3_key()],

                        c2.body(),

                    )

                    .unwrap();

                    CreateResponse::Created2(Created2::new(rep))

                }

            };

            created_send.send(reply).unwrap();

        };

        // Future to pretend to be a client.

        let client_fut = async move {

            let target = example_target();

            let params = CircParameters::default();

            let ret = match handshake_type {

                HandshakeType::Fast => {

                    trace!("doing fast create");

                    pending.create_firsthop_fast(params).await

                }

                HandshakeType::Ntor => {

                    trace!("doing ntor create");

                    pending.create_firsthop_ntor(&target, params).await

                }

                HandshakeType::NtorV3 => {

                    let params = if with_cc {

                        // Setup CC vegas parameters.

                        CircParameters::new(

                            true,

                            build_cc_vegas_params(),

                            FlowCtrlParameters::defaults_for_tests(),

                        )

                    } else {

                        params

                    };

                    trace!("doing ntor_v3 create");

                    pending.create_firsthop_ntor_v3(&target, params).await

                }

            };

            trace!("create done: result {:?}", ret);

            ret

        };

        let (circ, _) = futures::join!(client_fut, simulate_relay_fut);

        let _circ = circ.unwrap();

        // pfew!  We've build a circuit!  Let's make sure it has one hop.

        assert_eq!(_circ.n_hops().unwrap(), 1);

    }

    #[traced_test]

    #[test]

    fn test_create_fast() {

        tor_rtcompat::test_with_all_runtimes!(|rt| async move {

            test_create(&rt, HandshakeType::Fast, false).await;

        });

    }

    #[traced_test]

    #[test]

    fn test_create_ntor() {

        tor_rtcompat::test_with_all_runtimes!(|rt| async move {

            test_create(&rt, HandshakeType::Ntor, false).await;

        });

    }

    #[traced_test]

    #[test]

    fn test_create_ntor_v3() {

        tor_rtcompat::test_with_all_runtimes!(|rt| async move {

            test_create(&rt, HandshakeType::NtorV3, false).await;

        });

    }

    #[traced_test]

    #[test]

    #[cfg(feature = "flowctl-cc")]

    fn test_create_ntor_v3_with_cc() {

        tor_rtcompat::test_with_all_runtimes!(|rt| async move {

            test_create(&rt, HandshakeType::NtorV3, true).await;

        });

    }

    // An encryption layer that doesn't do any crypto.   Can be used

    // as inbound or outbound, but not both at once.

    pub(crate) struct DummyCrypto {

        counter_tag: [u8; 20],

        counter: u32,

        lasthop: bool,

    }

    impl DummyCrypto {

        fn next_tag(&mut self) -> SendmeTag {

            #![allow(clippy::identity_op)]

            self.counter_tag[0] = ((self.counter >> 0) & 255) as u8;

            self.counter_tag[1] = ((self.counter >> 8) & 255) as u8;