//! Handler for CREATE* cells.

use crate::FlowCtrlParameters;

use crate::ccparams::{

    AlgorithmDiscriminants, CongestionWindowParams, FixedWindowParams, RoundTripEstimatorParams,

    VegasParams,

};

use crate::channel::Channel;

use crate::circuit::celltypes::{CreateRequest, CreateResponse};

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

use crate::circuit::{

    CircuitRxSender, HandshakeSubprotocols, InvalidHandshakeSubprotocolError, UniqId,

};

use crate::client::circuit::padding::PaddingController;

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

use crate::crypto::cell::CryptInit as _;

use crate::crypto::cell::{

    CgoRelayCrypto, InboundRelayLayer, OutboundRelayLayer, RelayLayer, Tor1RelayCrypto,

};

use crate::crypto::handshake::RelayHandshakeError;

use crate::crypto::handshake::ServerHandshake as _;

use crate::crypto::handshake::fast::CreateFastServer;

use crate::crypto::handshake::ntor::{NtorSecretKey, NtorServer};

use crate::crypto::handshake::ntor_v3::{NtorV3SecretKey, NtorV3Server};

use crate::memquota::SpecificAccount as _;

use crate::memquota::{ChannelAccount, CircuitAccount};

use crate::relay::channel_provider::ChannelProvider;

use crate::relay::reactor::Reactor;

use crate::relay::{IncomingStreamRequestFilter, RelayCirc};

use crate::stream::IncomingStream;

use futures::channel::mpsc;

use futures::{SinkExt, Stream};

use smallvec::SmallVec;

use std::sync::{Arc, RwLock, Weak};

use tor_cell::chancell::ChanMsg as _;

use tor_cell::chancell::CircId;

use tor_cell::chancell::msg::{

    CreateFast, Created2, CreatedFast, Destroy, DestroyReason, HandshakeType,

};

use tor_cell::relaycell::RelayCmd;

use tor_cell::relaycell::extend::{

    CcRequest, CcResponse, CircRequestExt, CircResponseExt, SubprotocolRequest,

};

use tor_error::{ErrorKind, HasKind, debug_report, internal, into_internal, warn_report};

use tor_linkspec::OwnedChanTarget;

use tor_llcrypto::pk::ed25519::Ed25519Identity;

use tor_llcrypto::pk::rsa::RsaIdentity;

use tor_memquota::mq_queue::ChannelSpec as _;

use tor_memquota::mq_queue::MpscSpec;

use tor_relay_crypto::pk::{RelayNtorKeypair, RelayNtorKeys};

use tor_rtcompat::SpawnExt as _;

use tor_rtcompat::{DynTimeProvider, Runtime};

use tracing::{debug, trace};

/// Everything needed to handle CREATE* messages on channels.

#[derive(derive_more::Debug)]

pub struct CreateRequestHandler {

    /// Something that can launch channels. Typically the `ChanMgr`.

    chan_provider: Weak<dyn ChannelProvider<BuildSpec = OwnedChanTarget> + Send + Sync>,

    /// Circuit-related network parameters.

    circ_net_params: RwLock<CircNetParameters>,

    /// The circuit extension keys.

    #[debug(skip)]

    ntor_keys: RwLock<RelayNtorKeys>,

    /// An [`IncomingStreamRequestFilter`] factory for checking whether the user wants

    /// this request, or wants to reject it immediately.

    ///

    /// Used for obtaining a current [`IncomingStreamRequestFilter`]

    /// for building a circuit reactor.

    //

    // TODO(relay): it's likely this will end up changing quite a bit once we start

    // figuring out exactly how the config/reconfigure() logic and IncomingStreamRequestFilter

    // should function for relays.

    #[debug(skip)]

    incoming_filter_factory: Box<dyn IncomingStreamRequestFilterFactory + Send + Sync>,

    /// The allowed incoming stream commands.

    ///

    /// Used for rejecting BEGIN and RESOLVE if we are not configured to be an exit.

    ///

    // TODO(relay): we might use this for rejecting BEGIN_DIR too,

    // if we decide to allow relays to opt out of being dir mirrors.

    // See https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/4107/diffs#note_3426447

    allowed_stream_cmds: SmallVec<[RelayCmd; 3]>,

    /// A sender for the [`Stream`]s of `IncomingStream` of all circuits.

    ///

    /// The receiver will receive one [`Stream`] (of tor streams) per circuit.

    ///

    /// This being a bounded MPSC might seem a bit risky, because in theory,

    /// if the receiver is not reading fast enough, sending will block.

    /// In practice, however, it should never block (or buffer very much at all,

    /// for that matter), because the user (arti-relay) is expected to read from

    /// this in a tight loop, and spawn a task for handling each [`Stream`].

    ///

    /// Note: because this MPSC is not associated with any particular circuit or channel,

    /// it does not participate in the memquota system (see [crate::memquota]).

    #[debug(skip)]

    circuit_stream_tx: mpsc::Sender<Box<dyn Stream<Item = IncomingStream> + Send + Sync + Unpin>>,

}

// We make the CREATE-handling methods of `CreateRequestHandler` async

// since we expect that in the future we may want to offload the crypto to a worker thread.

#[expect(clippy::unused_async)]

impl CreateRequestHandler {

    /// Build a new [`CreateRequestHandler`], and a [`CircuitIncomingStreamReceiver`]

    /// for receiving new streams that are opened on any incoming circuits.

        // TODO(relay-tuning): this MPSC can be a bottleneck,

        // as all the channels on this relay will want to send one item on it

        // each time a new circuit is created.

        //

        // The value set here is a guesstimate.

        const CIRC_STREAM_BUF_SIZE: usize = 1024;

        // This is not associated with any particular circuit

        // (it is for *all* circuits), so it doesn't participate in memquota

        // (see circuit_stream_tx docs)

        #[allow(clippy::disallowed_methods)]

    /// Update the circuit parameters from a network consensus.

    /// Update the handler with a new set of circuit extension keys.

    ///

    /// This is called periodically by the relay key rotation task.

    /// Handle a CREATE* cell.

    ///

    /// This intentionally does not return a [`crate::Error`] so that we don't accidentally shut

    /// down the channel reactor when we really should be returning a DESTROY. Shutting down a

    /// channel may cause us to leak information about paths of circuits travelling through this

    /// relay. This is especially important here since we're handling data that is controllable from

    /// the other end of the circuit.

    #[allow(clippy::too_many_arguments)]

                // TODO(relay): The log messages throughout could be very noisy, so should have rate limiting.

                // `tor-spec/tearing-down-circuits.md`:

                //

                // > Implementations SHOULD always use the NONE reason to avoid side channels: [...]

            }

        }

    /// See [`Self::handle_create`].

    #[allow(clippy::too_many_arguments)]

        // Perform the handshake crypto and build the response.

                HandshakeType::NTOR_V3 => {

                }

                }

            },

        };

        // We use a large mpsc queue here since a circuit should never block the channel,

        // and we hope that memquota will help us if an attacker intentionally fills this buffer.

        // We use `10_000_000` since `usize::MAX` causes `futures::channel::mpsc` to panic.

        // TODO(relay): We should switch to an unbounded queue, but the circuit reactor is expecting

        // a bounded queue.

        // TODO(relay): Do we really want a client padding machine here?

        // Upgrade the channel provider, which in practice is the `ChanMgr` so this should not fail.

        };

        // Create an IncomingStreamRequestFilter for this circuit.

        // This will get applied to every stream request (BEGIN, BEGIN_DIR, RESOLVE)

        // arriving on the circuit.

        //

        // Note: once built, a circuit reactor's IncomingStreamRequestFilter cannot be changed

        // (it's fixed for the entire duration of the circuit).

        // Build the relay circuit reactor.

        )

        // Start the reactor in a task.

                // If we get here, it means the relay stream handler task has gone away,

                // so there won't be anything handling the incoming streams.

                //

                // The reactor is dropped, making the RelayCirc returned below

                // in the RelayCircComponents unusable

                // (RelayCirc::is_closing() will return `true`).

            } else {

                // Only spawn the circuit reactor if the incoming stream handler was

                // able to receive our message

                    }

                }

            }

    /// The handshake code for a CREATE_FAST request.

        // TODO(relay): We might want to offload this to a CPU worker in the future.

            // The CREATE_FAST handshake doesn't accept or return extensions,

            // so this `AuxDataReply` is a no-op.

            // The CREATE_FAST handshake doesn't use any keys.

        // No subprotocols are requested during a CREATE_FAST handshake.

            // CREATE_FAST always uses fixed-window flow control.

    /// The handshake code for a CREATE2 ntor (non-v3) request.

        // TODO(relay): We might want to offload this to a CPU worker in the future.

            // The ntor (non-v3) handshake doesn't accept or return extensions,

            // so this `AuxDataReply` is a no-op.

        // No subprotocols are requested during an ntor (non-v3) handshake.

            // CREATE2 with ntor (non-v3) always uses fixed-window flow control.

    /// The handshake code for a CREATE2 ntor-v3 request.

        // These extensions can be negotiated during the handshake.

        // These subprotocols were requested during the handshake.

        // They are not validated.

        // Helper which processes extension requests and returns any responses.

        // Returns `None` if the handshake should fail.

            // https://spec.torproject.org/tor-spec/create-created-cells.html#additional-data

            //

            // > Unless otherwise specified in the documentation for an extension type:

            // > - [...]

            // > - Parties MUST ignore any occurrence of an extension with a given type after the first such occurrence.

            //

            // TODO: Is there something nicer that we can do here?

            // We could use accessors like `ExtList::get_cc_request()`

            // which iterate over the extension list for each extension,

            // but using an enum match like we do below is kind of nice.

                    CircRequestExt::CcRequest(CcRequest { .. }) => {

                    }

                    // The given `SubprotocolRequest` stores a list of `NumberedSubver`,

                    // but a circuit extension request is limited to 255 bytes (127 subprotocols).

                    // So while a malicious client could send us a lot of invalid subprotocols,

                    // this limit prevents this list from being excessively large.

                        // We don't check the requested subprotocols here.

                    }

                        // https://spec.torproject.org/tor-spec/create-created-cells.html#additional-data

                        //

                        // > Parties MUST ignore extensions with `EXT_FIELD_TYPE` bodies they do not recognize.

                            ?ext,

                            "CREATE2 ntor-v3 handshake requested unrecognized extension",

                        );

                    }

                        // https://spec.torproject.org/tor-spec/create-created-cells.html#additional-data

                        //

                        // > Parties MUST ignore extensions with `EXT_FIELD_TYPE` bodies they do not recognize.

                        //

                        // We recognize this but don't know what to do with it.

                        // We haven't implemented it, or it doesn't make sense

                        // (for example `CircRequestExt::ProofOfWork`).

                        // So we'll just behave as if we don't recognize it.

                            ?ext,

                            "CREATE2 ntor-v3 handshake requested unsupported extension",

                        );

                    }

                }

            }

        // TODO(relay): We might want to offload this to a CPU worker in the future.

        // Ensure that the client did not request invalid/unsupported subprotocols.

        } else {

        };

    /// Helper to get the ntor keypairs after some transformation `map`.

    ///

    /// The `map` transformation must be fast since it blocks a read lock.

    /// The returned keys are sorted with the most recent key first.

    ///

    /// It would be nice if this just returned an iterator,

    /// but the read lock prevents this.

}

/// A receiver of [`Stream`]s (one for each incoming circuit),

/// where each `Stream` produces [`IncomingStream`]s for that circuit.

///

// Note: in theory, it would be nice if we could get rid of this type altogether.

// In an ideal world, I would've instead

//

//   * added a `RelayCirc::take_incoming_streams()` method for obtaining

//     the futures::Stream of IncomingStream of that circuit

//   * made the CreateRequestHandler send each Arc<RelayCirc> over to arti-relay for handling

//   * made arti-relay obtain the futures::Stream<Item = IncomingStream> of each RelayCirc

//     by calling `RelayCirc::take_incoming_streams()`

//

// However, that would involve adding some locking/interior mutability within RelayCirc

// (which is always behind an Arc), or extending mq_queue::Receiver to be Clone,

// which would be tricky to pull off (see the comment on mq_queue::Receiver about this).

pub struct CircuitIncomingStreamReceiver {

    /// The receiver for the [`Stream`]s of `IncomingStream` of all circuits.

    ///

    /// Receives one [`Stream`] (of tor streams) per circuit.

    /// Each of these will be handled in a new task.

    circuit_stream_rx: mpsc::Receiver<<Self as Stream>::Item>,

}

impl Stream for CircuitIncomingStreamReceiver {

    // TODO: it would be nice if we could return a type-erased Stream here

    // (impl Stream<...>), but impl Trait in associated types is unstable.

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

    type Item = Box<dyn Stream<Item = IncomingStream> + Send + Sync + Unpin>;

        use futures::StreamExt as _;

}

/// Helper function to split a `RelayLayer` into forward and backward type-erased trait objects.

{

/// An error that occurred while handling a CREATE* request.

#[derive(Debug, thiserror::Error)]

enum HandleCreateError {

    /// Circuit relay handshake failed.

    #[error("Circuit relay handshake failed")]

    Handshake(#[from] RelayHandshakeError),

    /// Circuit relay handshake failed.

    #[error("Failed to process the circuit relay handshake parameters")]

    HandshakeParameters(#[from] HandshakeParamsError),

    /// Requested subprotocols which aren't supported.

    #[error("Client requested subprotocol(s) which aren't supported")]

    HandshakeSubprotocols(#[from] InvalidHandshakeSubprotocolError),

    /// The requested handshake type is unsupported.

    #[error("Unsupported handshake type {0}")]

    Create2HandshakeType(HandshakeType),

    /// A memquota error.

    #[error("Memquota error")]

    Memquota(#[from] tor_memquota::Error),

    /// Error when spawning a task.

    #[error("Runtime task spawn error")]

    Spawn(#[from] futures::task::SpawnError),

    /// An internal error.

    ///

    /// Note that other variants (such as `Handshake` containing a [`RelayHandshakeError`])

    /// may themselves contain internal errors.

    #[error("Internal error")]

    Internal(#[from] tor_error::Bug),

}

impl HasKind for HandleCreateError {

        }

}

/// The components of a completed CREATE* handshake.

struct CompletedHandshakeComponents {

    /// The message to send in response.

    response: CreateResponse,

    /// The negotiated hop settings.

    hop_settings: HopSettings,

    /// Outbound onion crypto.

    crypto_out: Box<dyn OutboundRelayLayer + Send>,

    /// Inbound onion crypto.

    crypto_in: Box<dyn InboundRelayLayer + Send>,

}

/// A collection of objects built for a new relay circuit.

pub(crate) struct RelayCircComponents {

    /// The relay circuit handle.

    pub(crate) circ: Arc<RelayCirc>,

    /// Used to send data from the channel to the circuit reactor.

    pub(crate) sender: CircuitRxSender,

    /// The circuit's padding controller.

    pub(crate) padding_ctrl: PaddingController,

}

/// Congestion control network parameters.

#[derive(Debug, Clone)]

#[allow(clippy::exhaustive_structs)]

pub struct CongestionControlNetParams {

    /// Fixed-window algorithm parameters.

    pub fixed_window: FixedWindowParams,

    /// Vegas algorithm parameters for exit circuits.

    // NOTE: In this module we are handling CREATE* cells,

    // which only happens for non-hs circuits.

    // So we don't need to store the vegas hs parameters here.

    pub vegas_exit: VegasParams,

    /// Congestion window parameters.

    pub cwnd: CongestionWindowParams,

    /// RTT calculation parameters.

    pub rtt: RoundTripEstimatorParams,

    /// Flow control parameters to use for all streams on this circuit.

    pub flow_ctrl: FlowCtrlParameters,

}

impl CongestionControlNetParams {

    #[cfg(test)]

    // These have been copied from C-tor.

}

/// Network consensus parameters for handling incoming circuits.

///

/// Unlike `CircParameters`,

/// this is unopinionated and contains all relevant consensus parameters,

/// which is needed when handling an incoming CREATE* request where the

/// circuit origin chooses the type/settings

/// (for example congestion control type) of the circuit.

#[derive(Debug, Clone)]

#[allow(clippy::exhaustive_structs)]

pub struct CircNetParameters {

    /// Congestion control network parameters.

    pub cc: CongestionControlNetParams,

}

/// An [`IncomingStreamRequestFilter`] factory for building [`IncomingStreamRequestFilter`]s.

///

/// Each time a new circuit is opened, the [`CreateRequestHandler`] calls

/// [`IncomingStreamRequestFilterFactory::current_filter`] to build

/// an [`IncomingStreamRequestFilter`] for the circuit.

pub trait IncomingStreamRequestFilterFactory {

    /// Return the [`IncomingStreamRequestFilter`] to apply to the incoming stream requests

    /// arriving on a circuit.

    fn current_filter(&self) -> Box<dyn IncomingStreamRequestFilter>;

}

impl<F> IncomingStreamRequestFilterFactory for F

where

    F: Fn() -> Box<dyn IncomingStreamRequestFilter>,

{

}

#[cfg(test)]

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

    #![allow(clippy::string_slice)] // See arti#2571

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

    use tor_cell::chancell::msg::{AnyChanMsg, Create2, CreateFast, HandshakeType};

    use tor_cell::chancell::{AnyChanCell, ChanCmd, ChanMsg as _};

    use tor_rtcompat::test_with_one_runtime;

    use crate::channel::test_utils;

    use crate::circuit::CircParameters;

    /// Test the CREATE_FAST handshake.

    #[test]

    fn create_fast() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            let (client_chan, relay_chan, _circuit_stream_rx, _target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

            let circ_params = CircParameters::default();

            let tunnel = pending_tunnel

                .create_firsthop_fast(circ_params)

                .await

                .unwrap();

            // Client sent a CREATE_FAST, relay responded with a CREATED_FAST.

            assert_eq!(

                conn_inspector.try_client_cell().unwrap().msg().cmd(),

                ChanCmd::CREATE_FAST,

            );

            assert_eq!(

                conn_inspector.try_relay_cell().unwrap().msg().cmd(),

                ChanCmd::CREATED_FAST,

            );

            drop(tunnel);

            assert_eq!(

                conn_inspector.client_cell().await.unwrap().msg().cmd(),

                ChanCmd::DESTROY,

            );

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

    /// Test the CREATE_FAST handshake, but with a modified handshake payload.

    #[test]

    fn create_fast_fail() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            // Rewrite any client-sent CREATE_FAST messages to cause the client to fail the

            // handshake when processing the CREATED_FAST.

            conn_inspector.set_client_cell_modifier(|cell: &mut AnyChanCell| {

                let circ_id = cell.circid();

                if let AnyChanMsg::CreateFast(msg) = cell.msg() {

                    let mut new_handshake = msg.handshake().to_vec();

                    // Flip all bits.

                    for byte in &mut new_handshake {

                        *byte = !*byte;

                    }

                    // Reassemble the CREATE_FAST cell with the incorrect handshake body.

                    let new_msg = CreateFast::new(new_handshake);

                    *cell = AnyChanCell::new(circ_id, AnyChanMsg::CreateFast(new_msg));

                }

            });

            let (client_chan, relay_chan, _circuit_stream_rx, _target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

            let circ_params = CircParameters::default();

            // I don't think it's possible to modify the CREATE_FAST cell to make the relay fail the

            // handshake (the CREATE_FAST payload consists of only random bytes),

            // so the relay will respond successfully with a CREATED_FAST.

            // But the client will fail since the CREATED_FAST will contain garbage bytes.

            // The relay successfully processed the handshake,

            // but the client fails to validate the handshake since the handshake data was modified.

            assert!(matches!(

                pending_tunnel.create_firsthop_fast(circ_params).await,

                Err(crate::Error::BadCircHandshakeAuth),

            ));

            // Client sent a CREATE_FAST, relay responded with a CREATED_FAST.

            assert_eq!(

                conn_inspector.try_client_cell().unwrap().msg().cmd(),

                ChanCmd::CREATE_FAST,

            );

            assert_eq!(

                conn_inspector.try_relay_cell().unwrap().msg().cmd(),

                ChanCmd::CREATED_FAST,

            );

            // Since the `create_firsthop_fast()` failed above,

            // the client should have sent a DESTROY.

            assert_eq!(

                conn_inspector.client_cell().await.unwrap().msg().cmd(),

                ChanCmd::DESTROY,

            );

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

    /// Test the client with the "Relay=1" subprotocol (corresponds to the TAP handshake),

    /// which Arti doesn't support.

    #[test]

    fn tap() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            let (client_chan, _relay_chan, _circuit_stream_rx, mut target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

            let circ_params = CircParameters::default();

            // https://spec.torproject.org/tor-spec/subprotocol-versioning.html

            // 1 = RELAY_BASE

            let protocols = "Relay=1".parse().unwrap();

            let target = target_builder.protocols(protocols).build().unwrap();

            // TODO: This should fail since we don't support TAP handshakes.

            // But the channel will do an ntor handshake anyway even though it's not supported.

            // https://gitlab.torproject.org/tpo/core/arti/-/work_items/2489

            let _tunnel = pending_tunnel

                .create_firsthop(&target, circ_params)

                .await

                .unwrap();

            // TODO: As above, this is wrong.

            assert_eq!(

                conn_inspector.try_client_cell().unwrap().msg().cmd(),

                ChanCmd::CREATE2,

            );

            assert_eq!(

                conn_inspector.try_relay_cell().unwrap().msg().cmd(),

                ChanCmd::CREATED2,

            );

        });

    }

    /// Test the CREATE2 ntor handshake.

    #[test]

    fn ntor() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            let (client_chan, relay_chan, _circuit_stream_rx, mut target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            // https://spec.torproject.org/tor-spec/subprotocol-versioning.html

            // 2 = RELAY_NTOR

            // 3 = RELAY_EXTEND_IPv6

            for relay_version in [2, 3] {

                let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

                let circ_params = CircParameters::default();

                let protocols = format!("Relay=2-{relay_version}").parse().unwrap();

                let target = target_builder.protocols(protocols).build().unwrap();

                let tunnel = pending_tunnel

                    .create_firsthop(&target, circ_params)

                    .await

                    .unwrap();

                let client_cell = conn_inspector.try_client_cell().unwrap().msg().clone();

                let relay_cell = conn_inspector.try_relay_cell().unwrap().msg().clone();

                // Client sent a CREATE2, relay responded with a CREATED2.

                assert_eq!(client_cell.cmd(), ChanCmd::CREATE2);

                assert_eq!(relay_cell.cmd(), ChanCmd::CREATED2);

                // Check that it was an ntor handshake.

                let AnyChanMsg::Create2(client_cell) = client_cell else {

                    unreachable!("CREATE2 checked above");

                };

                assert_eq!(client_cell.handshake_type(), HandshakeType::NTOR);

                drop(tunnel);

                assert_eq!(

                    conn_inspector.client_cell().await.unwrap().msg().cmd(),

                    ChanCmd::DESTROY,

                );

                // We don't expect any other messages to have been sent.

                assert!(conn_inspector.try_client_cell().is_none());

                assert!(conn_inspector.try_relay_cell().is_none());

            }

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

    /// Test the CREATE2 ntor handshake, but with a modified handshake payload.

    #[test]

    fn ntor_fail() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            // Rewrite any client-sent CREATE2 messages to cause the relay to fail the handshake.

            conn_inspector.set_client_cell_modifier(|cell: &mut AnyChanCell| {

                let circ_id = cell.circid();

                if let AnyChanMsg::Create2(msg) = cell.msg() {

                    let mut new_body = msg.body().to_vec();

                    // Flip some arbitrarily chosen byte.

                    new_body[10] = !new_body[10];

                    // Reassemble the CREATE2 cell with the incorrect handshake body.

                    let new_msg = Create2::new(msg.handshake_type(), new_body);

                    *cell = AnyChanCell::new(circ_id, AnyChanMsg::Create2(new_msg));

                }

            });

            let (client_chan, relay_chan, _circuit_stream_rx, mut target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            // https://spec.torproject.org/tor-spec/subprotocol-versioning.html

            // 2 = RELAY_NTOR

            // 3 = RELAY_EXTEND_IPv6

            for relay_version in [2, 3] {

                let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

                let circ_params = CircParameters::default();

                let protocols = format!("Relay=2-{relay_version}").parse().unwrap();

                let target = target_builder.protocols(protocols).build().unwrap();

                // The relay refuses the circuit handshake since the CREATED2 cell had some

                // intentional issue with the handshake body.

                assert!(matches!(

                    pending_tunnel.create_firsthop(&target, circ_params).await,

                    Err(crate::Error::CircRefused(_)),

                ));

                // Client sent a CREATE2, relay responded with a DESTROY.

                assert_eq!(

                    conn_inspector.try_client_cell().unwrap().msg().cmd(),

                    ChanCmd::CREATE2,

                );

                assert_eq!(

                    conn_inspector.try_relay_cell().unwrap().msg().cmd(),

                    ChanCmd::DESTROY,

                );

                // We don't expect any other messages to have been sent.

                assert!(conn_inspector.try_client_cell().is_none());

                assert!(conn_inspector.try_relay_cell().is_none());

            }

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

    /// Test the CREATE2 ntor-v3 handshake.

    #[test]

    fn ntor_v3() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            let (client_chan, relay_chan, _circuit_stream_rx, mut target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            // https://spec.torproject.org/tor-spec/subprotocol-versioning.html

            // 4 = RELAY_NTORV3

            // 5 = RELAY_NEGOTIATE_SUBPROTO

            // 6 = RELAY_CRYPT_CGO

            for relay_version in [4, 5, 6] {

                let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

                let circ_params = CircParameters::default();

                let protocols = format!("Relay=4-{relay_version}").parse().unwrap();

                let target = target_builder.protocols(protocols).build().unwrap();

                let tunnel = pending_tunnel

                    .create_firsthop(&target, circ_params)

                    .await

                    .unwrap();

                let client_cell = conn_inspector.try_client_cell().unwrap().msg().clone();

                let relay_cell = conn_inspector.try_relay_cell().unwrap().msg().clone();

                // Client sent a CREATE2, relay responded with a CREATED2.

                assert_eq!(client_cell.cmd(), ChanCmd::CREATE2);

                assert_eq!(relay_cell.cmd(), ChanCmd::CREATED2);

                // Check that it was an ntor-v3 handshake.

                let AnyChanMsg::Create2(client_cell) = client_cell else {

                    unreachable!("CREATE2 checked above");

                };

                assert_eq!(client_cell.handshake_type(), HandshakeType::NTOR_V3);

                // TODO: It would be nice if we had a way to check that CGO was in use when

                // `relay_version` is >=6, but I don't see a nice way to do that.

                drop(tunnel);

                assert_eq!(

                    conn_inspector.client_cell().await.unwrap().msg().cmd(),

                    ChanCmd::DESTROY,

                );

                // We don't expect any other messages to have been sent.

                assert!(conn_inspector.try_client_cell().is_none());

                assert!(conn_inspector.try_relay_cell().is_none());

            }

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

    /// Test the CREATE2 ntor-v3 handshake, but with a modified handshake payload.

    #[test]

    fn ntor_v3_fail() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            // Rewrite any client-sent CREATE2 messages to cause the relay to fail the handshake.

            conn_inspector.set_client_cell_modifier(|cell: &mut AnyChanCell| {

                let circ_id = cell.circid();

                if let AnyChanMsg::Create2(msg) = cell.msg() {

                    let mut new_body = msg.body().to_vec();

                    // Flip some arbitrarily chosen byte.

                    new_body[10] = !new_body[10];

                    // Reassemble the CREATE2 cell with the incorrect handshake body.

                    let new_msg = Create2::new(msg.handshake_type(), new_body);

                    *cell = AnyChanCell::new(circ_id, AnyChanMsg::Create2(new_msg));

                }

            });

            let (client_chan, relay_chan, _circuit_stream_rx, mut target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            // https://spec.torproject.org/tor-spec/subprotocol-versioning.html

            // 4 = RELAY_NTORV3

            // 5 = RELAY_NEGOTIATE_SUBPROTO

            // 6 = RELAY_CRYPT_CGO

            for relay_version in [4, 5, 6] {

                let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

                let circ_params = CircParameters::default();

                let protocols = format!("Relay=4-{relay_version}").parse().unwrap();

                let target = target_builder.protocols(protocols).build().unwrap();

                // The relay refuses the circuit handshake since the CREATED2 cell had some

                // intentional issue with the handshake body.

                assert!(matches!(

                    pending_tunnel.create_firsthop(&target, circ_params).await,

                    Err(crate::Error::CircRefused(_)),

                ));

                // Client sent a CREATE2, relay responded with a DESTROY.

                assert_eq!(

                    conn_inspector.try_client_cell().unwrap().msg().cmd(),

                    ChanCmd::CREATE2,

                );

                assert_eq!(

                    conn_inspector.try_relay_cell().unwrap().msg().cmd(),

                    ChanCmd::DESTROY,

                );

                // We don't expect any other messages to have been sent.

                assert!(conn_inspector.try_client_cell().is_none());

                assert!(conn_inspector.try_relay_cell().is_none());

            }

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

    /// Test the CREATE2 handshake, but with an invalid handshake type.

    #[test]

    fn create2_invalid_handshake_fail() {

        test_with_one_runtime!(|rt| async move {

            let mut conn_inspector = test_utils::ConnInspector::new();

            // Rewrite any client-sent CREATE2 messages to cause the relay to fail the handshake.

            conn_inspector.set_client_cell_modifier(|cell: &mut AnyChanCell| {

                let circ_id = cell.circid();

                if let AnyChanMsg::Create2(msg) = cell.msg() {

                    // Reassemble the CREATE2 cell with the incorrect handshake type.

                    let new_msg = tor_cell::chancell::msg::Create2::new(99.into(), msg.body());

                    *cell = AnyChanCell::new(circ_id, AnyChanMsg::Create2(new_msg));

                }

            });

            let (client_chan, relay_chan, _circuit_stream_rx, mut target_builder) =

                test_utils::new_channel_pair_with_keys(&rt, &conn_inspector);

            // https://spec.torproject.org/tor-spec/subprotocol-versioning.html

            // 2 = RELAY_NTOR

            // 3 = RELAY_EXTEND_IPv6

            // 4 = RELAY_NTORV3

            // 5 = RELAY_NEGOTIATE_SUBPROTO

            // 6 = RELAY_CRYPT_CGO

            for relay_version in [2, 6] {

                let pending_tunnel = test_utils::new_pending_tunnel(&rt, &client_chan).await;

                let circ_params = CircParameters::default();

                let protocols = format!("Relay=2-{relay_version}").parse().unwrap();

                let target = target_builder.protocols(protocols).build().unwrap();

                // The relay refuses the circuit handshake since the CREATED2 cell had some

                // unrecognized handshake type.

                assert!(matches!(

                    pending_tunnel.create_firsthop(&target, circ_params).await,

                    Err(crate::Error::CircRefused(_)),

                ));

                // Client sent a CREATE2, relay responded with a DESTROY.

                assert_eq!(

                    conn_inspector.try_client_cell().unwrap().msg().cmd(),

                    ChanCmd::CREATE2,

                );

                assert_eq!(

                    conn_inspector.try_relay_cell().unwrap().msg().cmd(),

                    ChanCmd::DESTROY,

                );

                // We don't expect any other messages to have been sent.

                assert!(conn_inspector.try_client_cell().is_none());

                assert!(conn_inspector.try_relay_cell().is_none());

            }

            // Wait for both channels to close (ignoring any channel errors).

            let wait_fut =

                futures::future::join(client_chan.wait_for_close(), relay_chan.wait_for_close());

            drop((client_chan, relay_chan));

            let _ = wait_fut.await;

            // We don't expect any other messages to have been sent.

            assert!(conn_inspector.try_client_cell().is_none());

            assert!(conn_inspector.try_relay_cell().is_none());

        });

    }

}