//! Main implementation of the connection functionality

use std::time::Duration;

use std::collections::HashMap;

use std::fmt::Debug;

use std::marker::PhantomData;

use std::sync::Arc;

use std::time::Instant;

use async_trait::async_trait;

use educe::Educe;

use futures::{AsyncRead, AsyncWrite};

use itertools::Itertools;

use rand::Rng;

use tor_bytes::Writeable;

use tor_cell::relaycell::hs::intro_payload::{self, IntroduceHandshakePayload};

use tor_cell::relaycell::hs::pow::ProofOfWork;

use tor_cell::relaycell::msg::{AnyRelayMsg, Introduce1, Rendezvous2};

use tor_circmgr::build::onion_circparams_from_netparams;

use tor_circmgr::{

    ClientOnionServiceDataTunnel, ClientOnionServiceDirTunnel, ClientOnionServiceIntroTunnel,

};

use tor_dirclient::SourceInfo;

use tor_error::{Bug, debug_report, warn_report};

use tor_hscrypto::Subcredential;

use tor_proto::TargetHop;

use tor_proto::client::circuit::handshake::hs_ntor;

use tracing::{debug, instrument, trace};

use retry_error::RetryError;

use safelog::{DispRedacted, Sensitive};

use tor_cell::relaycell::RelayMsg;

use tor_cell::relaycell::hs::{

    AuthKeyType, EstablishRendezvous, IntroduceAck, RendezvousEstablished,

};

use tor_checkable::{Timebound, timed::TimerangeBound};

use tor_circmgr::hspool::HsCircPool;

use tor_circmgr::timeouts::Action as TimeoutsAction;

use tor_dirclient::request::Requestable as _;

use tor_error::{HasRetryTime as _, RetryTime};

use tor_error::{internal, into_internal};

use tor_hscrypto::RendCookie;

use tor_hscrypto::pk::{HsBlindId, HsId, HsIdKey};

use tor_linkspec::{CircTarget, HasRelayIds, OwnedCircTarget, RelayId};

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

use tor_netdir::{NetDir, Relay};

use tor_netdoc::doc::hsdesc::{HsDesc, IntroPointDesc};

use tor_proto::client::circuit::CircParameters;

use tor_proto::{MetaCellDisposition, MsgHandler};

use tor_rtcompat::{Runtime, SleepProviderExt as _, TimeoutError};

use crate::Config;

use crate::pow::HsPowClient;

use crate::proto_oneshot;

use crate::relay_info::ipt_to_circtarget;

use crate::state::MockableConnectorData;

use crate::{ConnError, DescriptorError, DescriptorErrorDetail};

use crate::{FailedAttemptError, IntroPtIndex, RendPtIdentityForError, rend_pt_identity_for_error};

use crate::{HsClientConnector, HsClientSecretKeys};

use ConnError as CE;

use FailedAttemptError as FAE;

/// Number of hops in our hsdir, introduction, and rendezvous circuits

///

/// Required by `tor_circmgr`'s timeout estimation API

/// ([`tor_circmgr::CircMgr::estimate_timeout`], [`HsCircPool::estimate_timeout`]).

///

/// TODO HS hardcoding the number of hops to 3 seems wrong.

/// This is really something that HsCircPool knows.  And some setups might want to make

/// shorter circuits for some reason.  And it will become wrong with vanguards?

/// But right now I think this is what HsCircPool does.

//

// Some commentary from

//   https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1342#note_2918050

// Possibilities:

//  * Look at n_hops() on the circuits we get, if we don't need this estimate

//    till after we have the circuit.

//  * Add a function to HsCircPool to tell us what length of circuit to expect

//    for each given type of circuit.

const HOPS: usize = 3;

/// Given `R, M` where `M: MocksForConnect<M>`, expand to the mockable `ClientCirc`

// This is quite annoying.  But the alternative is to write out `<... as // ...>`

// each time, since otherwise the compile complains about ambiguous associated types.

macro_rules! DataTunnel{ { $R:ty, $M:ty } => {

    <<$M as MocksForConnect<$R>>::HsCircPool as MockableCircPool<$R>>::DataTunnel

} }

/// Information about a hidden service, including our connection history

#[derive(Default, Educe)]

#[educe(Debug)]

// This type is actually crate-private, since it isn't re-exported, but it must

// be `pub` because it appears as a default for a type parameter in HsClientConnector.

pub struct Data {

    /// The latest known onion service descriptor for this service.

    desc: DataHsDesc,

    /// Information about the latest status of trying to connect to this service

    /// through each of its introduction points.

    ipts: DataIpts,

}

/// Part of `Data` that relates to the HS descriptor

type DataHsDesc = Option<TimerangeBound<HsDesc>>;

/// Part of `Data` that relates to our information about introduction points

type DataIpts = HashMap<RelayIdForExperience, IptExperience>;

/// How things went last time we tried to use this introduction point

///

/// Neither this data structure, nor [`Data`], is responsible for arranging that we expire this

/// information eventually.  If we keep reconnecting to the service, we'll retain information

/// about each IPT indefinitely, at least so long as they remain listed in the descriptors we

/// receive.

///

/// Expiry of unused data is handled by `state.rs`, according to `last_used` in `ServiceState`.

///

/// Choosing which IPT to prefer is done by obtaining an `IptSortKey`

/// (from this and other information).

//

// Don't impl Ord for IptExperience.  We obtain `Option<&IptExperience>` from our

// data structure, and if IptExperience were Ord then Option<&IptExperience> would be Ord

// but it would be the wrong sort order: it would always prefer None, ie untried IPTs.

#[derive(Debug)]

struct IptExperience {

    /// How long it took us to get whatever outcome occurred

    ///

    /// We prefer fast successes to slow ones.

    /// Then, we prefer failures with earlier `RetryTime`,

    /// and, lastly, faster failures to slower ones.

    duration: Duration,

    /// What happened and when we might try again

    ///

    /// Note that we don't actually *enforce* the `RetryTime` here, just sort by it

    /// using `RetryTime::loose_cmp`.

    ///

    /// We *do* return an error that is itself `HasRetryTime` and expect our callers

    /// to honour that.

    outcome: Result<(), RetryTime>,

}

/// Actually make a HS connection, updating our recorded state as necessary

///

/// `connector` is provided only for obtaining the runtime and netdir (and `mock_for_state`).

/// Obviously, `connect` is not supposed to go looking in `services`.

///

/// This function handles all necessary retrying of fallible operations,

/// (and, therefore, must also limit the total work done for a particular call).

///

/// This function has a minimum of functionality, since it is the boundary

/// between "mock connection, used for testing `state.rs`" and

/// "mock circuit and netdir, used for testing `connect.rs`",

/// so it is not, itself, unit-testable.

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

    Context::new(

        &connector.runtime,

        &*connector.circpool,

        netdir,

        config,

        hsid,

        secret_keys,

        (),

    )?

    .connect(data)

    .await

/// Common context for a single request to connect to a hidden service

///

/// This saves on passing this same set of (immutable) values (or subsets thereof)

/// to each method in the principal functional code, everywhere.

/// It also provides a convenient type to be `Self`.

///

/// Its lifetime is one request to make a new client circuit to a hidden service,

/// including all the retries and timeouts.

struct Context<'c, R: Runtime, M: MocksForConnect<R>> {

    /// Runtime

    runtime: &'c R,

    /// Circpool

    circpool: &'c M::HsCircPool,

    /// Netdir

    //

    // TODO holding onto the netdir for the duration of our attempts is not ideal

    // but doing better is fairly complicated.  See discussions here:

    //   https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1228#note_2910545

    //   https://gitlab.torproject.org/tpo/core/arti/-/issues/884

    netdir: Arc<NetDir>,

    /// Configuration

    config: Arc<Config>,

    /// Secret keys to use

    secret_keys: HsClientSecretKeys,

    /// HS ID

    hsid: DispRedacted<HsId>,

    /// Blinded HS ID

    hs_blind_id: HsBlindId,

    /// The subcredential to use during this time period

    subcredential: Subcredential,

    /// Mock data

    mocks: M,

}

/// Details of an established rendezvous point

///

/// Intermediate value for progress during a connection attempt.

struct Rendezvous<'r, R: Runtime, M: MocksForConnect<R>> {

    /// RPT as a `Relay`

    rend_relay: Relay<'r>,

    /// Rendezvous circuit

    rend_tunnel: DataTunnel!(R, M),

    /// Rendezvous cookie

    rend_cookie: RendCookie,

    /// Receiver that will give us the RENDEZVOUS2 message.

    ///

    /// The sending ended is owned by the handler

    /// which receives control messages on the rendezvous circuit,

    /// and which was installed when we sent `ESTABLISH_RENDEZVOUS`.

    ///

    /// (`RENDEZVOUS2` is the message containing the onion service's side of the handshake.)

    rend2_rx: proto_oneshot::Receiver<Rendezvous2>,

    /// Dummy, to placate compiler

    ///

    /// Covariant without dropck or interfering with Send/Sync will do fine.

    marker: PhantomData<fn() -> (R, M)>,

}

/// Random value used as part of IPT selection

type IptSortRand = u32;

/// Details of an apparently-useable introduction point

///

/// Intermediate value for progress during a connection attempt.

struct UsableIntroPt<'i> {

    /// Index in HS descriptor

    intro_index: IntroPtIndex,

    /// IPT descriptor

    intro_desc: &'i IntroPointDesc,

    /// IPT `CircTarget`

    intro_target: OwnedCircTarget,

    /// Random value used as part of IPT selection

    sort_rand: IptSortRand,

}

/// Lookup key for looking up and recording our IPT use experiences

///

/// Used to identify a relay when looking to see what happened last time we used it,

/// and storing that information after we tried it.

///

/// We store the experience information under an arbitrary one of the relay's identities,

/// as returned by the `HasRelayIds::identities().next()`.

/// When we do lookups, we check all the relay's identities to see if we find

/// anything relevant.

/// If relay identities permute in strange ways, whether we find our previous

/// knowledge about them is not particularly well defined, but that's fine.

///

/// While this is, structurally, a relay identity, it is not suitable for other purposes.

#[derive(Hash, Eq, PartialEq, Ord, PartialOrd, Debug)]

struct RelayIdForExperience(RelayId);

/// Details of an apparently-successful INTRODUCE exchange

///

/// Intermediate value for progress during a connection attempt.

struct Introduced<R: Runtime, M: MocksForConnect<R>> {

    /// End-to-end crypto NTORv3 handshake with the service

    ///

    /// Created as part of generating our `INTRODUCE1`,

    /// and then used when processing `RENDEZVOUS2`.

    handshake_state: hs_ntor::HsNtorClientState,

    /// Dummy, to placate compiler

    ///

    /// `R` and `M` only used for getting to mocks.

    /// Covariant without dropck or interfering with Send/Sync will do fine.

    marker: PhantomData<fn() -> (R, M)>,

}

impl RelayIdForExperience {

    /// Identities to use to try to find previous experience information about this IPT

    /// Identity to use to store previous experience information about this IPT

}

/// Sort key for an introduction point, for selecting the best IPTs to try first

///

/// Ordering is most preferable first.

///

/// We use this to sort our `UsableIpt`s using `.sort_by_key`.

/// (This implementation approach ensures that we obey all the usual ordering invariants.)

#[derive(Ord, PartialOrd, Eq, PartialEq, Debug)]

struct IptSortKey {

    /// Sort by how preferable the experience was

    outcome: IptSortKeyOutcome,

    /// Failing that, choose randomly

    sort_rand: IptSortRand,

}

/// Component of the [`IptSortKey`] representing outcome of our last attempt, if any

///

/// This is the main thing we use to decide which IPTs to try first.

/// It is calculated for each IPT

/// (via `.sort_by_key`, so repeatedly - it should therefore be cheap to make.)

///

/// Ordering is most preferable first.

#[derive(Ord, PartialOrd, Eq, PartialEq, Debug)]

enum IptSortKeyOutcome {

    /// Prefer successes

    Success {

        /// Prefer quick ones

        duration: Duration,

    },

    /// Failing that, try one we don't know to have failed

    Untried,

    /// Failing that, it'll have to be ones that didn't work last time

    Failed {

        /// Prefer failures with an earlier retry time

        retry_time: tor_error::LooseCmpRetryTime,

        /// Failing that, prefer quick failures (rather than slow ones eg timeouts)

        duration: Duration,

    },

}

impl From<Option<&IptExperience>> for IptSortKeyOutcome {

        use IptSortKeyOutcome as O;

            },

        }

}

impl<'c, R: Runtime, M: MocksForConnect<R>> Context<'c, R, M> {

    /// Make a new `Context` from the input data

                // TODO HS what on earth do these errors mean, in practical terms ?

                // In particular, we'll want to convert them to a ConnError variant,

                // but what ErrorKind should they have ?

    /// Actually make a HS connection, updating our recorded state as necessary

    ///

    /// Called by the `connect` function in this module.

    ///

    /// This function handles all necessary retrying of fallible operations,

    /// (and, therefore, must also limit the total work done for a particular call).

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

        // This function must do the following, retrying as appropriate.

        //  - Look up the onion descriptor in the state.

        //  - Download the onion descriptor if one isn't there.

        //  - In parallel:

        //    - Pick a rendezvous point from the netdirprovider and launch a

        //      rendezvous circuit to it. Then send ESTABLISH_INTRO.

        //    - Pick a number of introduction points (1 or more) and try to

        //      launch circuits to them.

        //  - On a circuit to an introduction point, send an INTRODUCE1 cell.

        //  - Wait for a RENDEZVOUS2 cell on the rendezvous circuit

        //  - Add a virtual hop to the rendezvous circuit.

        //  - Return the rendezvous circuit.

        let mocks = self.mocks.clone();

        let desc = self.descriptor_ensure(&mut data.desc).await?;

        mocks.test_got_desc(desc);

        let tunnel = self.intro_rend_connect(desc, &mut data.ipts).await?;

        mocks.test_got_tunnel(&tunnel);

        Ok(tunnel)

    /// Ensure that `Data.desc` contains the HS descriptor

    ///

    /// If we have a previously-downloaded descriptor, which is still valid,

    /// just returns a reference to it.

    ///

    /// Otherwise, tries to obtain the descriptor by downloading it from hsdir(s).

    ///

    /// Does all necessary retries and timeouts.

    /// Returns an error if no valid descriptor could be found.

    #[allow(clippy::cognitive_complexity)] // TODO: Refactor

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

        // Maximum number of hsdir connection and retrieval attempts we'll make

        let max_total_attempts = self

            .config

            .retry

            .hs_desc_fetch_attempts()

            .try_into()

            // User specified a very large u32.  We must be downcasting it to 16bit!

            // let's give them as many retries as we can manage.

            .unwrap_or(usize::MAX);

        // Limit on the duration of each retrieval attempt

        let each_timeout = self.estimate_timeout(&[

            (1, TimeoutsAction::BuildCircuit { length: HOPS }), // build circuit

            (1, TimeoutsAction::RoundTrip { length: HOPS }),    // One HTTP query/response

        ]);

        // We retain a previously obtained descriptor precisely until its lifetime expires,

        // and pay no attention to the descriptor's revision counter.

        // When it expires, we discard it completely and try to obtain a new one.

        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914448

        // TODO SPEC: Discuss HS descriptor lifetime and expiry client behaviour

            let now = self.runtime.wallclock();

            if let Ok(_desc) = previously.as_ref().check_valid_at(&now) {

                // Ideally we would just return desc but that confuses borrowck.

                // https://github.com/rust-lang/rust/issues/51545

                return Ok(data

                    .as_ref()

                    .expect("Some but now None")

                    .as_ref()

                    .check_valid_at(&now)

                    .expect("Ok but now Err"));

            }

            // Seems to be not valid now.  Try to fetch a fresh one.

        }

        let hs_dirs = self.netdir.hs_dirs_download(

            self.hs_blind_id,

            self.netdir.hs_time_period(),

            &mut self.mocks.thread_rng(),

        )?;

        trace!(

            "HS desc fetch for {}, using {} hsdirs",

            &self.hsid,

            hs_dirs.len()

        );

        // We might consider launching requests to multiple HsDirs in parallel.

        //   https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1118#note_2894463

        // But C Tor doesn't and our HS experts don't consider that important:

        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914436

        // (Additionally, making multiple HSDir requests at once may make us

        // more vulnerable to traffic analysis.)

        let mut attempts = hs_dirs.iter().cycle().take(max_total_attempts);

        let mut errors = RetryError::in_attempt_to("retrieve hidden service descriptor");

        let desc = loop {

            let relay = match attempts.next() {

                Some(relay) => relay,

                None => {

                    return Err(if errors.is_empty() {

                        CE::NoHsDirs

                    } else {

                        CE::DescriptorDownload(errors)

                    });

                }

            };

            let hsdir_for_error: Sensitive<Ed25519Identity> = (*relay.id()).into();

            match self

                .runtime

                .timeout(each_timeout, self.descriptor_fetch_attempt(relay))

                .await

                .unwrap_or(Err(DescriptorErrorDetail::Timeout))

            {

                Ok(desc) => break desc,

                Err(error) => {

                    if error.should_report_as_suspicious() {

                        // Note that not every protocol violation is suspicious:

                        // we only warn on the protocol violations that look like attempts

                        // to do a traffic tagging attack via hsdir inflation.

                        // (See proposal 360.)

                        warn_report!(

                            &error,

                            "Suspicious failure while downloading hsdesc for {} from relay {}",

                            &self.hsid,

                            relay.display_relay_ids(),

                        );

                    } else {

                        debug_report!(

                            &error,

                            "failed hsdir desc fetch for {} from {}/{}",

                            &self.hsid,

                            &relay.id(),

                            &relay.rsa_id()

                        );

                    }

                    errors.push_timed(

                        tor_error::Report(DescriptorError {

                            hsdir: hsdir_for_error,

                            error,

                        }),

                        self.runtime.now(),

                        Some(self.runtime.wallclock()),

                    );

                }

            }

        };

        // Store the bounded value in the cache for reuse,

        // but return a reference to the unwrapped `HsDesc`.

        //

        // The `HsDesc` must be owned by `data.desc`,

        // so first add it to `data.desc`,

        // and then dangerously_assume_timely to get a reference out again.

        //

        // It is safe to dangerously_assume_timely,

        // as descriptor_fetch_attempt has already checked the timeliness of the descriptor.

        let ret = data.insert(desc);

        Ok(ret.as_ref().dangerously_assume_timely())

    /// Make one attempt to fetch the descriptor from a specific hsdir

    ///

    /// No timeout

    ///

    /// On success, returns the descriptor.

    ///

    /// While the returned descriptor is `TimerangeBound`, its validity at the current time *has*

    /// been checked.

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

        let max_len: usize = self

            .netdir

            .params()

            .hsdir_max_desc_size

            .get()

            .try_into()

            .map_err(into_internal!("BoundedInt was not truly bounded!"))?;

        let request = {

            let mut r = tor_dirclient::request::HsDescDownloadRequest::new(self.hs_blind_id);

            r.set_max_len(max_len);

            r

        };

        trace!(

            "hsdir for {}, trying {}/{}, request {:?} (http request {:?})",

            &self.hsid,

            &hsdir.id(),

            &hsdir.rsa_id(),

            &request,

            request.debug_request()

        );

        let circuit = self

            .circpool

            .m_get_or_launch_dir(&self.netdir, OwnedCircTarget::from_circ_target(hsdir))

            .await?;

        let source: Option<SourceInfo> = circuit

            .m_source_info()

            .map_err(into_internal!("Couldn't get SourceInfo for circuit"))?;

        let mut stream = circuit

            .m_begin_dir_stream()

            .await

            .map_err(DescriptorErrorDetail::Circuit)?;

        let response = tor_dirclient::send_request(self.runtime, &request, &mut stream, source)

            .await

        let desc_text = response.into_output_string().map_err(|rfe| rfe.error)?;

        let hsc_desc_enc = self.secret_keys.keys.ks_hsc_desc_enc.as_ref();

        let now = self.runtime.wallclock();

        HsDesc::parse_decrypt_validate(

            &desc_text,

            &self.hs_blind_id,

            now,

            &self.subcredential,

            hsc_desc_enc,

        )

        .map_err(DescriptorErrorDetail::from)

    /// Given the descriptor, try to connect to service

    ///

    /// Does all necessary retries, timeouts, etc.

        // Maximum number of rendezvous/introduction attempts we'll make

            // User specified a very large u32.  We must be downcasting it to 16bit!

            // let's give them as many retries as we can manage.

        // Limit on the duration of each attempt to establish a rendezvous point

        //

        // This *might* include establishing a fresh circuit,

        // if the HsCircPool's pool is empty.

        // Limit on the duration of each attempt to negotiate with an introduction point

        //

        // *Does* include establishing the circuit.

        // Timeout estimator for the action that the HS will take in building

        // its circuit to the RPT.

            } else {

            },

        };

        // Limit on the duration of each attempt for activities involving both

        // RPT and IPT.

        // We can't reliably distinguish IPT failure from RPT failure, so we iterate over IPTs

        // (best first) and each time use a random RPT.

        // We limit the number of rendezvous establishment attempts, separately, since we don't

        // try to talk to the intro pt until we've established the rendezvous circuit.

        // But, we put all the errors into the same bucket, since we might have a mixture.

        // Note that IntroPtIndex is *not* the index into this Vec.

        // It is the index into the original list of introduction points in the descriptor.

                // Lack of TAIT means this clone

                }

        // Delete experience information for now-unlisted intro points

        // Otherwise, as the IPTs change `Data` might grow without bound,

        // if we keep reconnecting to the same HS.

        // Join with existing state recording our experiences,

        // sort by descending goodness, and then randomly

        // (so clients without any experience don't all pile onto the same, first, IPT)

        // We retain a rendezvous we managed to set up in here.  That way if we created it, and

        // then failed before we actually needed it, we can reuse it.

        // If we exit with an error, we will waste it - but because we isolate things we do

        // for different services, it wouldn't be reusable anyway.

        // If we are using proof-of-work DoS mitigation, this chooses an

        // algorithm and initial effort, and adjusts that effort when we retry.

        // We might consider making multiple INTRODUCE attempts to different

        // IPTs in in parallel, and somehow aggregating the errors and

        // experiences.

        // However our HS experts don't consider that important:

        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914438

        // Parallelizing our HsCircPool circuit building would likely have

        // greater impact. (See #1149.)

        loop {

            // When did we start doing things that depended on the IPT?

            //

            // Used for recording our experience with the selected IPT

            // Error handling inner async block (analogous to an IEFE):

            //  * Ok(Some()) means this attempt succeeded

            //  * Ok(None) means all attempts exhausted

            //  * Err(error) means this attempt failed

            //

            // Error handling is rather complex here.  It's the primary job of *this* code to

            // make sure that it's done right for timeouts.  (The individual component

            // functions handle non-timeout errors.)  The different timeout errors have

            // different amounts of information about the identity of the RPT and IPT: in each

            // case, the error only mentions the RPT or IPT if that node is implicated in the

            // timeout.

                // We establish a rendezvous point first.  Although it appears from reading

                // this code that this means we serialise establishment of the rendezvous and

                // introduction circuits, this isn't actually the case.  The circmgr maintains

                // a pool of circuits.  What actually happens in the "standing start" case is

                // that we obtain a circuit for rendezvous from the circmgr's pool, expecting

                // one to be available immediately; the circmgr will then start to build a new

                // one to replenish its pool, and that happens in parallel with the work we do

                // here - but in arrears.  If the circmgr pool is empty, then we must wait.

                //

                // Perhaps this should be parallelised here.  But that's really what the pool

                // is for, since we expect building the rendezvous circuit and building the

                // introduction circuit to take about the same length of time.

                //

                // We *do* serialise the ESTABLISH_RENDEZVOUS exchange, with the

                // building of the introduction circuit.  That could be improved, at the cost

                // of some additional complexity here.

                //

                // Our HS experts don't consider it important to increase the parallelism:

                //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914444

                //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914445

                    // Establish a rendezvous circuit.

                    };

                    );

                };

                            e

                        );

                    }

                };

                // We record how long things take, starting from here, as

                // as a statistic we'll use for the IPT in future.

                // This is stored in a variable outside this async block,

                // so that the outcome handling can use it.

                // No `Option::get_or_try_insert_with`, or we'd avoid this expect()

                );

                );

                        // The intro point ought to give us a prompt ACK regardless of HS

                        // behaviour or whatever is happening at the RPT, so blame the IPT.

                    // TODO: Maybe try, once, to extend-and-reuse the intro circuit.

                    //

                    // If the introduction fails, the introduction circuit is in principle

                    // still usable.  We believe that in this case, C Tor extends the intro

                    // circuit by one hop to the next IPT to try.  That saves on building a

                    // whole new 3-hop intro circuit.  However, our HS experts tell us that

                    // if introduction fails at one IPT it is likely to fail at the others too,

                    // so that optimisation might reduce our network impact and time to failure,

                    // but isn't likely to improve our chances of success.

                    //

                    // However, it's not clear whether this approach risks contaminating

                    // the 2nd attempt with some fault relating to the introduction point.

                    // The 1st ipt might also gain more knowledge about which HS we're talking to.

                    //

                    // TODO SPEC: Discuss extend-and-reuse HS intro circuit after nack

                #[allow(unused_variables)] // it's *supposed* to be unused

                    intro_index,

                );

            // Store the experience `outcome` we had with IPT `intro_index`, in `data`

            #[allow(clippy::unused_unit)] // -> () is here for error handling clarity

                })()

                    // Record successful outcome in Data

                }

                    // Record error outcome in Data, if in fact we involved the IPT

                    // at all.  The IPT information is be retrieved from `error`,

                    // since only some of the errors implicate the introduction point.

                    // If we are using proof-of-work DoS mitigation, try harder next time

                }

            }

        }

    /// Make one attempt to establish a rendezvous circuit

    ///

    /// This doesn't really depend on anything,

    /// other than (obviously) the isolation implied by our circuit pool.

    /// In particular it doesn't depend on the introduction point.

    ///

    /// Does not apply a timeout.

    ///

    /// On entry `using_rend_pt` is `None`.

    /// This function will store `Some` when it finds out which relay

    /// it is talking to and starts to converse with it.

    /// That way, if a timeout occurs, the caller can add that information to the error.

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

        let (rend_tunnel, rend_relay) = self

            .circpool

            .m_get_or_launch_client_rend(&self.netdir)

            .await

        let rend_pt = rend_pt_identity_for_error(&rend_relay);

        *using_rend_pt = Some(rend_pt.clone());

        let rend_cookie: RendCookie = self.mocks.thread_rng().random();

        let message = EstablishRendezvous::new(rend_cookie);

        let (rend_established_tx, rend_established_rx) = proto_oneshot::channel();

        let (rend2_tx, rend2_rx) = proto_oneshot::channel();

        /// Handler which expects `RENDEZVOUS_ESTABLISHED` and then

        /// `RENDEZVOUS2`.   Returns each message via the corresponding `oneshot`.

        struct Handler {

            /// Sender for a RENDEZVOUS_ESTABLISHED message.

            rend_established_tx: proto_oneshot::Sender<RendezvousEstablished>,

            /// Sender for a RENDEZVOUS2 message.

            rend2_tx: proto_oneshot::Sender<Rendezvous2>,

        }

        impl MsgHandler for Handler {

                // The first message we expect is a RENDEZVOUS_ESTABALISHED.

                } else {

                }

        }

        debug!(

            "hs conn to {}: RPT {}: sending ESTABLISH_RENDEZVOUS",

            &self.hsid,

            rend_pt.as_inner(),

        );

        let failed_map_err = |error| FAE::RendezvousEstablish {

        let handler = Handler {

            rend_established_tx,

            rend2_tx,

        };

        // TODO(conflux) This error handling is horrible. Problem is that this Mock system requires

        // to send back a tor_circmgr::Error while our reply handler requires a tor_proto::Error.

        // And unifying both is hard here considering it needs to be converted to yet another Error

        // type "FAE" so we have to do these hoops and jumps.

        rend_tunnel

            .m_start_conversation_last_hop(Some(message.into()), handler)

            .await

                };

        // `start_conversation` returns as soon as the control message has been sent.

        // We need to obtain the RENDEZVOUS_ESTABLISHED message, which is "returned" via the oneshot.

        let _: RendezvousEstablished = rend_established_rx.recv(failed_map_err).await?;

        debug!(

            "hs conn to {}: RPT {}: got RENDEZVOUS_ESTABLISHED",

            &self.hsid,

            rend_pt.as_inner(),

        );

        Ok(Rendezvous {

            rend_tunnel,

            rend_cookie,

            rend_relay,

            rend2_rx,

            marker: PhantomData,

        })

    /// Attempt (once) to send an INTRODUCE1 and wait for the INTRODUCE_ACK

    ///

    /// `take`s the input `rendezvous` (but only takes it if it gets that far)

    /// and, if successful, returns it.

    /// (This arranges that the rendezvous is "used up" precisely if

    /// we sent its secret somewhere.)

    ///

    /// Although this function handles the `Rendezvous`,

    /// nothing in it actually involves the rendezvous point.

    /// So if there's a failure, it's purely to do with the introduction point.

    ///

    /// Does not apply a timeout.

    #[allow(clippy::cognitive_complexity, clippy::type_complexity)] // TODO: Refactor

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

        debug!(

            "hs conn to {}: IPT {}: obtaining intro circuit",

            &self.hsid, intro_index,

        );

        let intro_circ = self

            .circpool

            .m_get_or_launch_intro(

                &self.netdir,

                ipt.intro_target.clone(), // &OwnedCircTarget isn't CircTarget apparently

            )

            .await

        let rend_pt = rend_pt_identity_for_error(&rendezvous.rend_relay);

        debug!(

            "hs conn to {}: RPT {} IPT {}: making introduction",

            &self.hsid,

            rend_pt.as_inner(),

            intro_index,

        );

        // Now we construct an introduce1 message and perform the first part of the

        // rendezvous handshake.

        //

        // This process is tricky because the header of the INTRODUCE1 message

        // -- which depends on the IntroPt configuration -- is authenticated as

        // part of the HsDesc handshake.

        // Construct the header, since we need it as input to our encryption.

        let intro_header = {

            let ipt_sid_key = ipt.intro_desc.ipt_sid_key();

            let intro1 = Introduce1::new(

                AuthKeyType::ED25519_SHA3_256,

                ipt_sid_key.as_bytes().to_vec(),

                vec![],

            );

            let mut header = vec![];

            intro1

                .encode_onto(&mut header)

                .map_err(into_internal!("couldn't encode intro1 header"))?;

            header

        };

        // Construct the introduce payload, which tells the onion service how to find

        // our rendezvous point.  (We could do this earlier if we wanted.)

        let intro_payload = {

            let onion_key =

                intro_payload::OnionKey::NtorOnionKey(*rendezvous.rend_relay.ntor_onion_key());

            let linkspecs = rendezvous

                .rend_relay

                .linkspecs()

                .map_err(into_internal!("Couldn't encode link specifiers"))?;

            let payload = IntroduceHandshakePayload::new(

                rendezvous.rend_cookie,

                onion_key,

                linkspecs,

                proof_of_work,

            );

            let mut encoded = vec![];

            payload

                .write_onto(&mut encoded)

                .map_err(into_internal!("Couldn't encode introduce1 payload"))?;

            encoded

        };

        // Perform the cryptographic handshake with the onion service.

        let service_info = hs_ntor::HsNtorServiceInfo::new(

            ipt.intro_desc.svc_ntor_key().clone(),

            ipt.intro_desc.ipt_sid_key().clone(),

            self.subcredential,

        );

        let handshake_state =

            hs_ntor::HsNtorClientState::new(&mut self.mocks.thread_rng(), service_info);

        let encrypted_body = handshake_state

            .client_send_intro(&intro_header, &intro_payload)

            .map_err(into_internal!("can't begin hs-ntor handshake"))?;

        // Build our actual INTRODUCE1 message.

        let intro1_real = Introduce1::new(

            AuthKeyType::ED25519_SHA3_256,

            ipt.intro_desc.ipt_sid_key().as_bytes().to_vec(),

            encrypted_body,

        );

        /// Handler which expects just `INTRODUCE_ACK`

        struct Handler {

            /// Sender for `INTRODUCE_ACK`

            intro_ack_tx: proto_oneshot::Sender<IntroduceAck>,

        }

        impl MsgHandler for Handler {

        }

        let (intro_ack_tx, intro_ack_rx) = proto_oneshot::channel();

        let handler = Handler { intro_ack_tx };

        debug!(

            "hs conn to {}: RPT {} IPT {}: making introduction - sending INTRODUCE1",

            &self.hsid,

            rend_pt.as_inner(),

            intro_index,

        );

        // TODO(conflux) This error handling is horrible. Problem is that this Mock system requires

        // to send back a tor_circmgr::Error while our reply handler requires a tor_proto::Error.

        // And unifying both is hard here considering it needs to be converted to yet another Error

        // type "FAE" so we have to do these hoops and jumps.

        intro_circ

            .m_start_conversation_last_hop(Some(intro1_real.into()), handler)

            .await

                };

        // Status is checked by `.success()`, and we don't look at the extensions;

        // just discard the known-successful `IntroduceAck`

        let _: IntroduceAck =

            intro_ack_rx

                .recv(failed_map_err)

                .await?

                .success()

                .map_err(|status| FAE::IntroductionFailed {

        debug!(

            "hs conn to {}: RPT {} IPT {}: making introduction - success",

            &self.hsid,

            rend_pt.as_inner(),

            intro_index,

        );

        // Having received INTRODUCE_ACK. we can forget about this circuit

        // (and potentially tear it down).

        drop(intro_circ);

        Ok((

            rendezvous,

            Introduced {

                handshake_state,

                marker: PhantomData,

            },

        ))

    /// Attempt (once) to connect a rendezvous circuit using the given intro pt

    ///

    /// Timeouts here might be due to the IPT, RPT, service,

    /// or any of the intermediate relays.

    ///

    /// If, rather than a timeout, we actually encounter some kind of error,

    /// we'll return the appropriate `FailedAttemptError`.

    /// (Who is responsible may vary, so the `FailedAttemptError` variant will reflect that.)

    ///

    /// Does not apply a timeout

        use tor_proto::client::circuit::handshake;

            intro_index,

        );

            intro_index,

        );

        // In theory would be great if we could have multiple introduction attempts in parallel

        // with similar x,X values but different IPTs.  However, our HS experts don't

        // think increasing parallelism here is important:

        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914438

        // Try to complete the cryptographic handshake.

            // If this goes wrong. either the onion service has mangled the crypto,

            // or the rendezvous point has misbehaved (that that is possible is a protocol bug),

            // or we have used the wrong handshake_state (let's assume that's not true).

            //

            // If this happens we'll go and try another RPT.