1
//! Main implementation of the connection functionality
2

            
3
use std::time::Duration;
4

            
5
use std::collections::HashMap;
6
use std::fmt::Debug;
7
use std::marker::PhantomData;
8
use std::sync::Arc;
9
use std::time::Instant;
10

            
11
use async_trait::async_trait;
12
use educe::Educe;
13
use futures::{AsyncRead, AsyncWrite};
14
use itertools::Itertools;
15
use rand::Rng;
16
use tor_bytes::Writeable;
17
use tor_cell::relaycell::hs::intro_payload::{self, IntroduceHandshakePayload};
18
use tor_cell::relaycell::msg::{AnyRelayMsg, Introduce1, Rendezvous2};
19
use tor_error::{debug_report, warn_report, Bug};
20
use tor_hscrypto::Subcredential;
21
use tor_proto::circuit::handshake::hs_ntor;
22
use tracing::{debug, trace};
23

            
24
use retry_error::RetryError;
25
use safelog::Sensitive;
26
use tor_cell::relaycell::hs::{
27
    AuthKeyType, EstablishRendezvous, IntroduceAck, RendezvousEstablished,
28
};
29
use tor_cell::relaycell::RelayMsg;
30
use tor_checkable::{timed::TimerangeBound, Timebound};
31
use tor_circmgr::build::circparameters_from_netparameters;
32
use tor_circmgr::hspool::{HsCircKind, HsCircPool};
33
use tor_circmgr::timeouts::Action as TimeoutsAction;
34
use tor_dirclient::request::Requestable as _;
35
use tor_error::{internal, into_internal};
36
use tor_error::{HasRetryTime as _, RetryTime};
37
use tor_hscrypto::pk::{HsBlindId, HsId, HsIdKey};
38
use tor_hscrypto::RendCookie;
39
use tor_linkspec::{CircTarget, HasRelayIds, OwnedCircTarget, RelayId};
40
use tor_llcrypto::pk::ed25519::Ed25519Identity;
41
use tor_netdir::{NetDir, Relay};
42
use tor_netdoc::doc::hsdesc::{HsDesc, IntroPointDesc};
43
use tor_proto::circuit::{
44
    CircParameters, ClientCirc, ConversationInHandler, MetaCellDisposition, MsgHandler,
45
};
46
use tor_rtcompat::{Runtime, SleepProviderExt as _, TimeoutError};
47

            
48
use crate::proto_oneshot;
49
use crate::relay_info::ipt_to_circtarget;
50
use crate::state::MockableConnectorData;
51
use crate::Config;
52
use crate::{rend_pt_identity_for_error, FailedAttemptError, IntroPtIndex, RendPtIdentityForError};
53
use crate::{ConnError, DescriptorError, DescriptorErrorDetail};
54
use crate::{HsClientConnector, HsClientSecretKeys};
55

            
56
use ConnError as CE;
57
use FailedAttemptError as FAE;
58

            
59
/// Number of hops in our hsdir, introduction, and rendezvous circuits
60
///
61
/// Required by `tor_circmgr`'s timeout estimation API
62
/// ([`tor_circmgr::CircMgr::estimate_timeout`], [`HsCircPool::estimate_timeout`]).
63
///
64
/// TODO HS hardcoding the number of hops to 3 seems wrong.
65
/// This is really something that HsCircPool knows.  And some setups might want to make
66
/// shorter circuits for some reason.  And it will become wrong with vanguards?
67
/// But right now I think this is what HsCircPool does.
68
//
69
// Some commentary from
70
//   https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1342#note_2918050
71
// Possibilities:
72
//  * Look at n_hops() on the circuits we get, if we don't need this estimate
73
//    till after we have the circuit.
74
//  * Add a function to HsCircPool to tell us what length of circuit to expect
75
//    for each given type of circuit.
76
const HOPS: usize = 3;
77

            
78
/// Given `R, M` where `M: MocksForConnect<M>`, expand to the mockable `ClientCirc`
79
// This is quite annoying.  But the alternative is to write out `<... as // ...>`
80
// each time, since otherwise the compile complains about ambiguous associated types.
81
macro_rules! ClientCirc { { $R:ty, $M:ty } => {
82
    <<$M as MocksForConnect<$R>>::HsCircPool as MockableCircPool<$R>>::ClientCirc
83
} }
84

            
85
/// Information about a hidden service, including our connection history
86
#[derive(Default, Educe)]
87
#[educe(Debug)]
88
// This type is actually crate-private, since it isn't re-exported, but it must
89
// be `pub` because it appears as a default for a type parameter in HsClientConnector.
90
pub struct Data {
91
    /// The latest known onion service descriptor for this service.
92
    desc: DataHsDesc,
93
    /// Information about the latest status of trying to connect to this service
94
    /// through each of its introduction points.
95
    ipts: DataIpts,
96
}
97

            
98
/// Part of `Data` that relates to the HS descriptor
99
type DataHsDesc = Option<TimerangeBound<HsDesc>>;
100

            
101
/// Part of `Data` that relates to our information about introduction points
102
type DataIpts = HashMap<RelayIdForExperience, IptExperience>;
103

            
104
/// How things went last time we tried to use this introduction point
105
///
106
/// Neither this data structure, nor [`Data`], is responsible for arranging that we expire this
107
/// information eventually.  If we keep reconnecting to the service, we'll retain information
108
/// about each IPT indefinitely, at least so long as they remain listed in the descriptors we
109
/// receive.
110
///
111
/// Expiry of unused data is handled by `state.rs`, according to `last_used` in `ServiceState`.
112
///
113
/// Choosing which IPT to prefer is done by obtaining an `IptSortKey`
114
/// (from this and other information).
115
//
116
// Don't impl Ord for IptExperience.  We obtain `Option<&IptExperience>` from our
117
// data structure, and if IptExperience were Ord then Option<&IptExperience> would be Ord
118
// but it would be the wrong sort order: it would always prefer None, ie untried IPTs.
119
#[derive(Debug)]
120
struct IptExperience {
121
    /// How long it took us to get whatever outcome occurred
122
    ///
123
    /// We prefer fast successes to slow ones.
124
    /// Then, we prefer failures with earlier `RetryTime`,
125
    /// and, lastly, faster failures to slower ones.
126
    duration: Duration,
127

            
128
    /// What happened and when we might try again
129
    ///
130
    /// Note that we don't actually *enforce* the `RetryTime` here, just sort by it
131
    /// using `RetryTime::loose_cmp`.
132
    ///
133
    /// We *do* return an error that is itself `HasRetryTime` and expect our callers
134
    /// to honour that.
135
    outcome: Result<(), RetryTime>,
136
}
137

            
138
/// Actually make a HS connection, updating our recorded state as necessary
139
///
140
/// `connector` is provided only for obtaining the runtime and netdir (and `mock_for_state`).
141
/// Obviously, `connect` is not supposed to go looking in `services`.
142
///
143
/// This function handles all necessary retrying of fallible operations,
144
/// (and, therefore, must also limit the total work done for a particular call).
145
///
146
/// This function has a minimum of functionality, since it is the boundary
147
/// between "mock connection, used for testing `state.rs`" and
148
/// "mock circuit and netdir, used for testing `connect.rs`",
149
/// so it is not, itself, unit-testable.
150
pub(crate) async fn connect<R: Runtime>(
151
    connector: &HsClientConnector<R>,
152
    netdir: Arc<NetDir>,
153
    config: Arc<Config>,
154
    hsid: HsId,
155
    data: &mut Data,
156
    secret_keys: HsClientSecretKeys,
157
) -> Result<Arc<ClientCirc>, ConnError> {
158
    Context::new(
159
        &connector.runtime,
160
        &*connector.circpool,
161
        netdir,
162
        config,
163
        hsid,
164
        secret_keys,
165
        (),
166
    )?
167
    .connect(data)
168
    .await
169
}
170

            
171
/// Common context for a single request to connect to a hidden service
172
///
173
/// This saves on passing this same set of (immutable) values (or subsets thereof)
174
/// to each method in the principal functional code, everywhere.
175
/// It also provides a convenient type to be `Self`.
176
///
177
/// Its lifetime is one request to make a new client circuit to a hidden service,
178
/// including all the retries and timeouts.
179
struct Context<'c, R: Runtime, M: MocksForConnect<R>> {
180
    /// Runtime
181
    runtime: &'c R,
182
    /// Circpool
183
    circpool: &'c M::HsCircPool,
184
    /// Netdir
185
    //
186
    // TODO holding onto the netdir for the duration of our attempts is not ideal
187
    // but doing better is fairly complicated.  See discussions here:
188
    //   https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1228#note_2910545
189
    //   https://gitlab.torproject.org/tpo/core/arti/-/issues/884
190
    netdir: Arc<NetDir>,
191
    /// Configuration
192
    config: Arc<Config>,
193
    /// Secret keys to use
194
    secret_keys: HsClientSecretKeys,
195
    /// HS ID
196
    hsid: HsId,
197
    /// Blinded HS ID
198
    hs_blind_id: HsBlindId,
199
    /// The subcredential to use during this time period
200
    subcredential: Subcredential,
201
    /// Mock data
202
    mocks: M,
203
}
204

            
205
/// Details of an established rendezvous point
206
///
207
/// Intermediate value for progress during a connection attempt.
208
struct Rendezvous<'r, R: Runtime, M: MocksForConnect<R>> {
209
    /// RPT as a `Relay`
210
    rend_relay: Relay<'r>,
211
    /// Rendezvous circuit
212
    rend_circ: Arc<ClientCirc!(R, M)>,
213
    /// Rendezvous cookie
214
    rend_cookie: RendCookie,
215

            
216
    /// Receiver that will give us the RENDEZVOUS2 message.
217
    ///
218
    /// The sending ended is owned by the handler
219
    /// which receives control messages on the rendezvous circuit,
220
    /// and which was installed when we sent `ESTABLISH_RENDEZVOUS`.
221
    ///
222
    /// (`RENDEZVOUS2` is the message containing the onion service's side of the handshake.)
223
    rend2_rx: proto_oneshot::Receiver<Rendezvous2>,
224

            
225
    /// Dummy, to placate compiler
226
    ///
227
    /// Covariant without dropck or interfering with Send/Sync will do fine.
228
    marker: PhantomData<fn() -> (R, M)>,
229
}
230

            
231
/// Random value used as part of IPT selection
232
type IptSortRand = u32;
233

            
234
/// Details of an apparently-useable introduction point
235
///
236
/// Intermediate value for progress during a connection attempt.
237
struct UsableIntroPt<'i> {
238
    /// Index in HS descriptor
239
    intro_index: IntroPtIndex,
240
    /// IPT descriptor
241
    intro_desc: &'i IntroPointDesc,
242
    /// IPT `CircTarget`
243
    intro_target: OwnedCircTarget,
244
    /// Random value used as part of IPT selection
245
    sort_rand: IptSortRand,
246
}
247

            
248
/// Lookup key for looking up and recording our IPT use experiences
249
///
250
/// Used to identify a relay when looking to see what happened last time we used it,
251
/// and storing that information after we tried it.
252
///
253
/// We store the experience information under an arbitrary one of the relay's identities,
254
/// as returned by the `HasRelayIds::identities().next()`.
255
/// When we do lookups, we check all the relay's identities to see if we find
256
/// anything relevant.
257
/// If relay identities permute in strange ways, whether we find our previous
258
/// knowledge about them is not particularly well defined, but that's fine.
259
///
260
/// While this is, structurally, a relay identity, it is not suitable for other purposes.
261
#[derive(Hash, Eq, PartialEq, Ord, PartialOrd, Debug)]
262
struct RelayIdForExperience(RelayId);
263

            
264
/// Details of an apparently-successful INTRODUCE exchange
265
///
266
/// Intermediate value for progress during a connection attempt.
267
struct Introduced<R: Runtime, M: MocksForConnect<R>> {
268
    /// End-to-end crypto NTORv3 handshake with the service
269
    ///
270
    /// Created as part of generating our `INTRODUCE1`,
271
    /// and then used when processing `RENDEZVOUS2`.
272
    handshake_state: hs_ntor::HsNtorClientState,
273

            
274
    /// Dummy, to placate compiler
275
    ///
276
    /// `R` and `M` only used for getting to mocks.
277
    /// Covariant without dropck or interfering with Send/Sync will do fine.
278
    marker: PhantomData<fn() -> (R, M)>,
279
}
280

            
281
impl RelayIdForExperience {
282
    /// Identities to use to try to find previous experience information about this IPT
283
    fn for_lookup(intro_target: &OwnedCircTarget) -> impl Iterator<Item = Self> + '_ {
284
        intro_target
285
            .identities()
286
            .map(|id| RelayIdForExperience(id.to_owned()))
287
    }
288

            
289
    /// Identity to use to store previous experience information about this IPT
290
    fn for_store(intro_target: &OwnedCircTarget) -> Result<Self, Bug> {
291
        let id = intro_target
292
            .identities()
293
            .next()
294
            .ok_or_else(|| internal!("introduction point relay with no identities"))?
295
            .to_owned();
296
        Ok(RelayIdForExperience(id))
297
    }
298
}
299

            
300
/// Sort key for an introduction point, for selecting the best IPTs to try first
301
///
302
/// Ordering is most preferable first.
303
///
304
/// We use this to sort our `UsableIpt`s using `.sort_by_key`.
305
/// (This implementation approach ensures that we obey all the usual ordering invariants.)
306
#[derive(Ord, PartialOrd, Eq, PartialEq, Debug)]
307
struct IptSortKey {
308
    /// Sort by how preferable the experience was
309
    outcome: IptSortKeyOutcome,
310
    /// Failing that, choose randomly
311
    sort_rand: IptSortRand,
312
}
313

            
314
/// Component of the [`IptSortKey`] representing outcome of our last attempt, if any
315
///
316
/// This is the main thing we use to decide which IPTs to try first.
317
/// It is calculated for each IPT
318
/// (via `.sort_by_key`, so repeatedly - it should therefore be cheap to make.)
319
///
320
/// Ordering is most preferable first.
321
#[derive(Ord, PartialOrd, Eq, PartialEq, Debug)]
322
enum IptSortKeyOutcome {
323
    /// Prefer successes
324
    Success {
325
        /// Prefer quick ones
326
        duration: Duration,
327
    },
328
    /// Failing that, try one we don't know to have failed
329
    Untried,
330
    /// Failing that, it'll have to be ones that didn't work last time
331
    Failed {
332
        /// Prefer failures with an earlier retry time
333
        retry_time: tor_error::LooseCmpRetryTime,
334
        /// Failing that, prefer quick failures (rather than slow ones eg timeouts)
335
        duration: Duration,
336
    },
337
}
338

            
339
impl From<Option<&IptExperience>> for IptSortKeyOutcome {
340
    fn from(experience: Option<&IptExperience>) -> IptSortKeyOutcome {
341
        use IptSortKeyOutcome as O;
342
        match experience {
343
            None => O::Untried,
344
            Some(IptExperience { duration, outcome }) => match outcome {
345
                Ok(()) => O::Success {
346
                    duration: *duration,
347
                },
348
                Err(retry_time) => O::Failed {
349
                    retry_time: (*retry_time).into(),
350
                    duration: *duration,
351
                },
352
            },
353
        }
354
    }
355
}
356

            
357
impl<'c, R: Runtime, M: MocksForConnect<R>> Context<'c, R, M> {
358
    /// Make a new `Context` from the input data
359
    fn new(
360
        runtime: &'c R,
361
        circpool: &'c M::HsCircPool,
362
        netdir: Arc<NetDir>,
363
        config: Arc<Config>,
364
        hsid: HsId,
365
        secret_keys: HsClientSecretKeys,
366
        mocks: M,
367
    ) -> Result<Self, ConnError> {
368
        let time_period = netdir.hs_time_period();
369
        let (hs_blind_id_key, subcredential) = HsIdKey::try_from(hsid)
370
            .map_err(|_| CE::InvalidHsId)?
371
            .compute_blinded_key(time_period)
372
            .map_err(
373
                // TODO HS what on earth do these errors mean, in practical terms ?
374
                // In particular, we'll want to convert them to a ConnError variant,
375
                // but what ErrorKind should they have ?
376
                into_internal!("key blinding error, don't know how to handle"),
377
            )?;
378
        let hs_blind_id = hs_blind_id_key.id();
379

            
380
        Ok(Context {
381
            netdir,
382
            config,
383
            hsid,
384
            hs_blind_id,
385
            subcredential,
386
            circpool,
387
            runtime,
388
            secret_keys,
389
            mocks,
390
        })
391
    }
392

            
393
    /// Actually make a HS connection, updating our recorded state as necessary
394
    ///
395
    /// Called by the `connect` function in this module.
396
    ///
397
    /// This function handles all necessary retrying of fallible operations,
398
    /// (and, therefore, must also limit the total work done for a particular call).
399
    async fn connect(&self, data: &mut Data) -> Result<Arc<ClientCirc!(R, M)>, ConnError> {
400
        // This function must do the following, retrying as appropriate.
401
        //  - Look up the onion descriptor in the state.
402
        //  - Download the onion descriptor if one isn't there.
403
        //  - In parallel:
404
        //    - Pick a rendezvous point from the netdirprovider and launch a
405
        //      rendezvous circuit to it. Then send ESTABLISH_INTRO.
406
        //    - Pick a number of introduction points (1 or more) and try to
407
        //      launch circuits to them.
408
        //  - On a circuit to an introduction point, send an INTRODUCE1 cell.
409
        //  - Wait for a RENDEZVOUS2 cell on the rendezvous circuit
410
        //  - Add a virtual hop to the rendezvous circuit.
411
        //  - Return the rendezvous circuit.
412

            
413
        let mocks = self.mocks.clone();
414

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

            
417
        mocks.test_got_desc(desc);
418

            
419
        let circ = self.intro_rend_connect(desc, &mut data.ipts).await?;
420
        mocks.test_got_circ(&circ);
421

            
422
        Ok(circ)
423
    }
424

            
425
    /// Ensure that `Data.desc` contains the HS descriptor
426
    ///
427
    /// If we have a previously-downloaded descriptor, which is still valid,
428
    /// just returns a reference to it.
429
    ///
430
    /// Otherwise, tries to obtain the descriptor by downloading it from hsdir(s).
431
    ///
432
    /// Does all necessary retries and timeouts.
433
    /// Returns an error if no valid descriptor could be found.
434
    async fn descriptor_ensure<'d>(&self, data: &'d mut DataHsDesc) -> Result<&'d HsDesc, CE> {
435
        // Maximum number of hsdir connection and retrieval attempts we'll make
436
        let max_total_attempts = self
437
            .config
438
            .retry
439
            .hs_desc_fetch_attempts()
440
            .try_into()
441
            // User specified a very large u32.  We must be downcasting it to 16bit!
442
            // let's give them as many retries as we can manage.
443
            .unwrap_or(usize::MAX);
444

            
445
        // Limit on the duration of each retrieval attempt
446
        let each_timeout = self.estimate_timeout(&[
447
            (1, TimeoutsAction::BuildCircuit { length: HOPS }), // build circuit
448
            (1, TimeoutsAction::RoundTrip { length: HOPS }),    // One HTTP query/response
449
        ]);
450

            
451
        // We retain a previously obtained descriptor precisely until its lifetime expires,
452
        // and pay no attention to the descriptor's revision counter.
453
        // When it expires, we discard it completely and try to obtain a new one.
454
        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914448
455
        // TODO SPEC: Discuss HS descriptor lifetime and expiry client behaviour
456
        if let Some(previously) = data {
457
            let now = self.runtime.wallclock();
458
            if let Ok(_desc) = previously.as_ref().check_valid_at(&now) {
459
                // Ideally we would just return desc but that confuses borrowck.
460
                // https://github.com/rust-lang/rust/issues/51545
461
                return Ok(data
462
                    .as_ref()
463
                    .expect("Some but now None")
464
                    .as_ref()
465
                    .check_valid_at(&now)
466
                    .expect("Ok but now Err"));
467
            }
468
            // Seems to be not valid now.  Try to fetch a fresh one.
469
        }
470

            
471
        let hs_dirs = self.netdir.hs_dirs_download(
472
            self.hs_blind_id,
473
            self.netdir.hs_time_period(),
474
            &mut self.mocks.thread_rng(),
475
        )?;
476

            
477
        trace!(
478
            "HS desc fetch for {}, using {} hsdirs",
479
            &self.hsid,
480
            hs_dirs.len()
481
        );
482

            
483
        // We might consider launching requests to multiple HsDirs in parallel.
484
        //   https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1118#note_2894463
485
        // But C Tor doesn't and our HS experts don't consider that important:
486
        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914436
487
        // (Additionally, making multiple HSDir requests at once may make us
488
        // more vulnerable to traffic analysis.)
489
        let mut attempts = hs_dirs.iter().cycle().take(max_total_attempts);
490
        let mut errors = RetryError::in_attempt_to("retrieve hidden service descriptor");
491
        let desc = loop {
492
            let relay = match attempts.next() {
493
                Some(relay) => relay,
494
                None => {
495
                    return Err(if errors.is_empty() {
496
                        CE::NoHsDirs
497
                    } else {
498
                        CE::DescriptorDownload(errors)
499
                    })
500
                }
501
            };
502
            let hsdir_for_error: Sensitive<Ed25519Identity> = (*relay.id()).into();
503
            match self
504
                .runtime
505
                .timeout(each_timeout, self.descriptor_fetch_attempt(relay))
506
                .await
507
                .unwrap_or(Err(DescriptorErrorDetail::Timeout))
508
            {
509
                Ok(desc) => break desc,
510
                Err(error) => {
511
                    debug_report!(
512
                        &error,
513
                        "failed hsdir desc fetch for {} from {}",
514
                        &self.hsid,
515
                        &relay.id(),
516
                    );
517
                    errors.push(tor_error::Report(DescriptorError {
518
                        hsdir: hsdir_for_error,
519
                        error,
520
                    }));
521
                }
522
            }
523
        };
524

            
525
        // Store the bounded value in the cache for reuse,
526
        // but return a reference to the unwrapped `HsDesc`.
527
        //
528
        // The `HsDesc` must be owned by `data.desc`,
529
        // so first add it to `data.desc`,
530
        // and then dangerously_assume_timely to get a reference out again.
531
        //
532
        // It is safe to dangerously_assume_timely,
533
        // as descriptor_fetch_attempt has already checked the timeliness of the descriptor.
534
        let ret = data.insert(desc);
535
        Ok(ret.as_ref().dangerously_assume_timely())
536
    }
537

            
538
    /// Make one attempt to fetch the descriptor from a specific hsdir
539
    ///
540
    /// No timeout
541
    ///
542
    /// On success, returns the descriptor.
543
    ///
544
    /// While the returned descriptor is `TimerangeBound`, its validity at the current time *has*
545
    /// been checked.
546
    async fn descriptor_fetch_attempt(
547
        &self,
548
        hsdir: &Relay<'_>,
549
    ) -> Result<TimerangeBound<HsDesc>, DescriptorErrorDetail> {
550
        let max_len: usize = self
551
            .netdir
552
            .params()
553
            .hsdir_max_desc_size
554
            .get()
555
            .try_into()
556
            .map_err(into_internal!("BoundedInt was not truly bounded!"))?;
557
        let request = {
558
            let mut r = tor_dirclient::request::HsDescDownloadRequest::new(self.hs_blind_id);
559
            r.set_max_len(max_len);
560
            r
561
        };
562
        trace!(
563
            "hsdir for {}, trying {}/{}, request {:?} (http request {:?})",
564
            &self.hsid,
565
            &hsdir.id(),
566
            &hsdir.rsa_id(),
567
            &request,
568
            request.debug_request()
569
        );
570

            
571
        let circuit = self
572
            .circpool
573
            .m_get_or_launch_specific(
574
                &self.netdir,
575
                HsCircKind::ClientHsDir,
576
                OwnedCircTarget::from_circ_target(hsdir),
577
            )
578
            .await?;
579
        let mut stream = circuit
580
            .m_begin_dir_stream()
581
            .await
582
            .map_err(DescriptorErrorDetail::Stream)?;
583

            
584
        let response = tor_dirclient::send_request(self.runtime, &request, &mut stream, None)
585
            .await
586
            .map_err(|dir_error| match dir_error {
587
                tor_dirclient::Error::RequestFailed(rfe) => DescriptorErrorDetail::from(rfe.error),
588
                tor_dirclient::Error::CircMgr(ce) => into_internal!(
589
                    "tor-dirclient complains about circmgr going wrong but we gave it a stream"
590
                )(ce)
591
                .into(),
592
                other => into_internal!(
593
                    "tor-dirclient gave unexpected error, tor-hsclient code needs updating"
594
                )(other)
595
                .into(),
596
            })?;
597

            
598
        let desc_text = response.into_output_string().map_err(|rfe| rfe.error)?;
599
        let hsc_desc_enc = self.secret_keys.keys.ks_hsc_desc_enc.as_ref();
600

            
601
        let now = self.runtime.wallclock();
602

            
603
        HsDesc::parse_decrypt_validate(
604
            &desc_text,
605
            &self.hs_blind_id,
606
            now,
607
            &self.subcredential,
608
            hsc_desc_enc,
609
        )
610
        .map_err(DescriptorErrorDetail::from)
611
    }
612

            
613
    /// Given the descriptor, try to connect to service
614
    ///
615
    /// Does all necessary retries, timeouts, etc.
616
    async fn intro_rend_connect(
617
        &self,
618
        desc: &HsDesc,
619
        data: &mut DataIpts,
620
    ) -> Result<Arc<ClientCirc!(R, M)>, CE> {
621
        // Maximum number of rendezvous/introduction attempts we'll make
622
        let max_total_attempts = self
623
            .config
624
            .retry
625
            .hs_intro_rend_attempts()
626
            .try_into()
627
            // User specified a very large u32.  We must be downcasting it to 16bit!
628
            // let's give them as many retries as we can manage.
629
            .unwrap_or(usize::MAX);
630

            
631
        // Limit on the duration of each attempt to establish a rendezvous point
632
        //
633
        // This *might* include establishing a fresh circuit,
634
        // if the HsCircPool's pool is empty.
635
        let rend_timeout = self.estimate_timeout(&[
636
            (1, TimeoutsAction::BuildCircuit { length: HOPS }), // build circuit
637
            (1, TimeoutsAction::RoundTrip { length: HOPS }),    // One ESTABLISH_RENDEZVOUS
638
        ]);
639

            
640
        // Limit on the duration of each attempt to negotiate with an introduction point
641
        //
642
        // *Does* include establishing the circuit.
643
        let intro_timeout = self.estimate_timeout(&[
644
            (1, TimeoutsAction::BuildCircuit { length: HOPS }), // build circuit
645
            // This does some crypto too, but we don't account for that.
646
            (1, TimeoutsAction::RoundTrip { length: HOPS }), // One INTRODUCE1/INTRODUCE_ACK
647
        ]);
648

            
649
        // Limit on the duration of each attempt for activities involving both RPT and IPT
650
        let hs_hops = if desc.is_single_onion_service() {
651
            1
652
        } else {
653
            HOPS
654
        };
655
        let rpt_ipt_timeout = self.estimate_timeout(&[
656
            // The API requires us to specify a number of circuit builds and round trips.
657
            // So what we tell the estimator is a rather imprecise description.
658
            // (TODO it would be nice if the circmgr offered us a one-way trip Action).
659
            //
660
            // What we are timing here is:
661
            //
662
            //    INTRODUCE2 goes from IPT to HS
663
            //    but that happens in parallel with us waiting for INTRODUCE_ACK,
664
            //    which is controlled by `intro_timeout` so not pat of `ipt_rpt_timeout`.
665
            //    and which has to come HOPS hops.  So don't count INTRODUCE2 here.
666
            //
667
            //    HS builds to our RPT
668
            (1, TimeoutsAction::BuildCircuit { length: hs_hops }),
669
            //
670
            //    RENDEZVOUS1 goes from HS to RPT.  `hs_hops`, one-way.
671
            //    RENDEZVOUS2 goes from RPT to us.  HOPS, one-way.
672
            //    Together, we squint a bit and call this a HOPS round trip:
673
            (1, TimeoutsAction::RoundTrip { length: HOPS }),
674
        ]);
675

            
676
        // We can't reliably distinguish IPT failure from RPT failure, so we iterate over IPTs
677
        // (best first) and each time use a random RPT.
678

            
679
        // We limit the number of rendezvous establishment attempts, separately, since we don't
680
        // try to talk to the intro pt until we've established the rendezvous circuit.
681
        let mut rend_attempts = 0..max_total_attempts;
682

            
683
        // But, we put all the errors into the same bucket, since we might have a mixture.
684
        let mut errors = RetryError::in_attempt_to("make circuit to to hidden service");
685

            
686
        // Note that IntroPtIndex is *not* the index into this Vec.
687
        // It is the index into the original list of introduction points in the descriptor.
688
        let mut usable_intros: Vec<UsableIntroPt> = desc
689
            .intro_points()
690
            .iter()
691
            .enumerate()
692
            .map(|(intro_index, intro_desc)| {
693
                let intro_index = intro_index.into();
694
                let intro_target = ipt_to_circtarget(intro_desc, &self.netdir)
695
                    .map_err(|error| FAE::UnusableIntro { error, intro_index })?;
696
                // Lack of TAIT means this clone
697
                let intro_target = OwnedCircTarget::from_circ_target(&intro_target);
698
                Ok::<_, FailedAttemptError>(UsableIntroPt {
699
                    intro_index,
700
                    intro_desc,
701
                    intro_target,
702
                    sort_rand: self.mocks.thread_rng().gen(),
703
                })
704
            })
705
            .filter_map(|entry| match entry {
706
                Ok(y) => Some(y),
707
                Err(e) => {
708
                    errors.push(e);
709
                    None
710
                }
711
            })
712
            .collect_vec();
713

            
714
        // Delete experience information for now-unlisted intro points
715
        // Otherwise, as the IPTs change `Data` might grow without bound,
716
        // if we keep reconnecting to the same HS.
717
        data.retain(|k, _v| {
718
            usable_intros
719
                .iter()
720
                .any(|ipt| RelayIdForExperience::for_lookup(&ipt.intro_target).any(|id| &id == k))
721
        });
722

            
723
        // Join with existing state recording our experiences,
724
        // sort by descending goodness, and then randomly
725
        // (so clients without any experience don't all pile onto the same, first, IPT)
726
        usable_intros.sort_by_key(|ipt: &UsableIntroPt| {
727
            let experience =
728
                RelayIdForExperience::for_lookup(&ipt.intro_target).find_map(|id| data.get(&id));
729
            IptSortKey {
730
                outcome: experience.into(),
731
                sort_rand: ipt.sort_rand,
732
            }
733
        });
734
        self.mocks.test_got_ipts(&usable_intros);
735

            
736
        let mut intro_attempts = usable_intros.iter().cycle().take(max_total_attempts);
737

            
738
        // We retain a rendezvous we managed to set up in here.  That way if we created it, and
739
        // then failed before we actually needed it, we can reuse it.
740
        // If we exit with an error, we will waste it - but because we isolate things we do
741
        // for different services, it wouldn't be reusable anyway.
742
        let mut saved_rendezvous = None;
743

            
744
        // We might consider making multiple INTRODUCE attempts to different
745
        // IPTs in in parallel, and somehow aggregating the errors and
746
        // experiences.
747
        // However our HS experts don't consider that important:
748
        //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914438
749
        // Parallelizing our HsCircPool circuit building would likely have
750
        // greater impact. (See #1149.)
751
        loop {
752
            // When did we start doing things that depended on the IPT?
753
            //
754
            // Used for recording our experience with the selected IPT
755
            let mut ipt_use_started = None::<Instant>;
756

            
757
            // Error handling inner async block (analogous to an IEFE):
758
            //  * Ok(Some()) means this attempt succeeded
759
            //  * Ok(None) means all attempts exhausted
760
            //  * Err(error) means this attempt failed
761
            //
762
            // Error handling is rather complex here.  It's the primary job of *this* code to
763
            // make sure that it's done right for timeouts.  (The individual component
764
            // functions handle non-timeout errors.)  The different timeout errors have
765
            // different amounts of information about the identity of the RPT and IPT: in each
766
            // case, the error only mentions the RPT or IPT if that node is implicated in the
767
            // timeout.
768
            let outcome = async {
769
                // We establish a rendezvous point first.  Although it appears from reading
770
                // this code that this means we serialise establishment of the rendezvous and
771
                // introduction circuits, this isn't actually the case.  The circmgr maintains
772
                // a pool of circuits.  What actually happens in the "standing start" case is
773
                // that we obtain a circuit for rendezvous from the circmgr's pool, expecting
774
                // one to be available immediately; the circmgr will then start to build a new
775
                // one to replenish its pool, and that happens in parallel with the work we do
776
                // here - but in arrears.  If the circmgr pool is empty, then we must wait.
777
                //
778
                // Perhaps this should be parallelised here.  But that's really what the pool
779
                // is for, since we expect building the rendezvous circuit and building the
780
                // introduction circuit to take about the same length of time.
781
                //
782
                // We *do* serialise the ESTABLISH_RENDEZVOUS exchange, with the
783
                // building of the introduction circuit.  That could be improved, at the cost
784
                // of some additional complexity here.
785
                //
786
                // Our HS experts don't consider it important to increase the parallelism:
787
                //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914444
788
                //   https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914445
789
                if saved_rendezvous.is_none() {
790
                    debug!("hs conn to {}: setting up rendezvous point", &self.hsid);
791
                    // Establish a rendezvous circuit.
792
                    let Some(_): Option<usize> = rend_attempts.next() else {
793
                        return Ok(None);
794
                    };
795

            
796
                    let mut using_rend_pt = None;
797
                    saved_rendezvous = Some(
798
                        self.runtime
799
                            .timeout(rend_timeout, self.establish_rendezvous(&mut using_rend_pt))
800
                            .await
801
                            .map_err(|_: TimeoutError| match using_rend_pt {
802
                                None => FAE::RendezvousCircuitObtain {
803
                                    error: tor_circmgr::Error::CircTimeout,
804
                                },
805
                                Some(rend_pt) => FAE::RendezvousEstablishTimeout { rend_pt },
806
                            })??,
807
                    );
808
                }
809

            
810
                let Some(ipt) = intro_attempts.next() else {
811
                    return Ok(None);
812
                };
813
                let intro_index = ipt.intro_index;
814

            
815
                // We record how long things take, starting from here, as
816
                // as a statistic we'll use for the IPT in future.
817
                // This is stored in a variable outside this async block,
818
                // so that the outcome handling can use it.
819
                ipt_use_started = Some(self.runtime.now());
820

            
821
                // No `Option::get_or_try_insert_with`, or we'd avoid this expect()
822
                let rend_pt_for_error = rend_pt_identity_for_error(
823
                    &saved_rendezvous
824
                        .as_ref()
825
                        .expect("just made Some")
826
                        .rend_relay,
827
                );
828
                debug!(
829
                    "hs conn to {}: RPT {}",
830
                    &self.hsid,
831
                    rend_pt_for_error.as_inner()
832
                );
833

            
834
                let (rendezvous, introduced) = self
835
                    .runtime
836
                    .timeout(
837
                        intro_timeout,
838
                        self.exchange_introduce(ipt, &mut saved_rendezvous),
839
                    )
840
                    .await
841
                    .map_err(|_: TimeoutError| {
842
                        // The intro point ought to give us a prompt ACK regardless of HS
843
                        // behaviour or whatever is happening at the RPT, so blame the IPT.
844
                        FAE::IntroductionTimeout { intro_index }
845
                    })?
846
                    // TODO: Maybe try, once, to extend-and-reuse the intro circuit.
847
                    //
848
	            // If the introduction fails, the introduction circuit is in principle
849
                    // still usable.  We believe that in this case, C Tor extends the intro
850
	            // circuit by one hop to the next IPT to try.  That saves on building a
851
                    // whole new 3-hop intro circuit.  However, our HS experts tell us that
852
                    // if introduction fails at one IPT it is likely to fail at the others too,
853
                    // so that optimisation might reduce our network impact and time to failure,
854
                    // but isn't likely to improve our chances of success.
855
                    //
856
                    // However, it's not clear whether this approach risks contaminating
857
                    // the 2nd attempt with some fault relating to the introduction point.
858
                    // The 1st ipt might also gain more knowledge about which HS we're talking to.
859
                    //
860
                    // TODO SPEC: Discuss extend-and-reuse HS intro circuit after nack
861
                    ?;
862
                #[allow(unused_variables)] // it's *supposed* to be unused
863
                let saved_rendezvous = (); // don't use `saved_rendezvous` any more, use rendezvous
864

            
865
                let rend_pt = rend_pt_identity_for_error(&rendezvous.rend_relay);
866
                let circ = self
867
                    .runtime
868
                    .timeout(
869
                        rpt_ipt_timeout,
870
                        self.complete_rendezvous(ipt, rendezvous, introduced),
871
                    )
872
                    .await
873
                    .map_err(|_: TimeoutError| FAE::RendezvousCompletionTimeout {
874
                        intro_index,
875
                        rend_pt: rend_pt.clone(),
876
                    })??;
877

            
878
                debug!(
879
                    "hs conn to {}: RPT {} IPT {}: success",
880
                    &self.hsid,
881
                    rend_pt.as_inner(),
882
                    intro_index,
883
                );
884
                Ok::<_, FAE>(Some((intro_index, circ)))
885
            }
886
            .await;
887

            
888
            // Store the experience `outcome` we had with IPT `intro_index`, in `data`
889
            #[allow(clippy::unused_unit)] // -> () is here for error handling clarity
890
            let mut store_experience = |intro_index, outcome| -> () {
891
                (|| {
892
                    let ipt = usable_intros
893
                        .iter()
894
                        .find(|ipt| ipt.intro_index == intro_index)
895
                        .ok_or_else(|| internal!("IPT not found by index"))?;
896
                    let id = RelayIdForExperience::for_store(&ipt.intro_target)?;
897
                    let started = ipt_use_started.ok_or_else(|| {
898
                        internal!("trying to record IPT use but no IPT start time noted")
899
                    })?;
900
                    let duration = self
901
                        .runtime
902
                        .now()
903
                        .checked_duration_since(started)
904
                        .ok_or_else(|| internal!("clock overflow calculating IPT use duration"))?;
905
                    data.insert(id, IptExperience { duration, outcome });
906
                    Ok::<_, Bug>(())
907
                })()
908
                .unwrap_or_else(|e| warn_report!(e, "error recording HS IPT use experience"));
909
            };
910

            
911
            match outcome {
912
                Ok(Some((intro_index, y))) => {
913
                    // Record successful outcome in Data
914
                    store_experience(intro_index, Ok(()));
915
                    return Ok(y);
916
                }
917
                Ok(None) => return Err(CE::Failed(errors)),
918
                Err(error) => {
919
                    debug_report!(&error, "hs conn to {}: attempt failed", &self.hsid);
920
                    // Record error outcome in Data, if in fact we involved the IPT
921
                    // at all.  The IPT information is be retrieved from `error`,
922
                    // since only some of the errors implicate the introduction point.
923
                    if let Some(intro_index) = error.intro_index() {
924
                        store_experience(intro_index, Err(error.retry_time()));
925
                    }
926
                    errors.push(error);
927
                }
928
            }
929
        }
930
    }
931

            
932
    /// Make one attempt to establish a rendezvous circuit
933
    ///
934
    /// This doesn't really depend on anything,
935
    /// other than (obviously) the isolation implied by our circuit pool.
936
    /// In particular it doesn't depend on the introduction point.
937
    ///
938
    /// Does not apply a timeout.
939
    ///
940
    /// On entry `using_rend_pt` is `None`.
941
    /// This function will store `Some` when it finds out which relay
942
    /// it is talking to and starts to converse with it.
943
    /// That way, if a timeout occurs, the caller can add that information to the error.
944
    async fn establish_rendezvous(
945
        &'c self,
946
        using_rend_pt: &mut Option<RendPtIdentityForError>,
947
    ) -> Result<Rendezvous<R, M>, FAE> {
948
        let (rend_circ, rend_relay) = self
949
            .circpool
950
            .m_get_or_launch_client_rend(&self.netdir)
951
            .await
952
            .map_err(|error| FAE::RendezvousCircuitObtain { error })?;
953

            
954
        let rend_pt = rend_pt_identity_for_error(&rend_relay);
955
        *using_rend_pt = Some(rend_pt.clone());
956

            
957
        let rend_cookie: RendCookie = self.mocks.thread_rng().gen();
958
        let message = EstablishRendezvous::new(rend_cookie);
959

            
960
        let (rend_established_tx, rend_established_rx) = proto_oneshot::channel();
961
        let (rend2_tx, rend2_rx) = proto_oneshot::channel();
962

            
963
        /// Handler which expects `RENDEZVOUS_ESTABLISHED` and then
964
        /// `RENDEZVOUS2`.   Returns each message via the corresponding `oneshot`.
965
        struct Handler {
966
            /// Sender for a RENDEZVOUS_ESTABLISHED message.
967
            rend_established_tx: proto_oneshot::Sender<RendezvousEstablished>,
968
            /// Sender for a RENDEZVOUS2 message.
969
            rend2_tx: proto_oneshot::Sender<Rendezvous2>,
970
        }
971
        impl MsgHandler for Handler {
972
            fn handle_msg(
973
                &mut self,
974
                _conversation: ConversationInHandler<'_, '_, '_>,
975
                msg: AnyRelayMsg,
976
            ) -> Result<MetaCellDisposition, tor_proto::Error> {
977
                // The first message we expect is a RENDEZVOUS_ESTABALISHED.
978
                if self.rend_established_tx.still_expected() {
979
                    self.rend_established_tx
980
                        .deliver_expected_message(msg, MetaCellDisposition::Consumed)
981
                } else {
982
                    self.rend2_tx
983
                        .deliver_expected_message(msg, MetaCellDisposition::ConversationFinished)
984
                }
985
            }
986
        }
987

            
988
        debug!(
989
            "hs conn to {}: RPT {}: sending ESTABLISH_RENDEZVOUS",
990
            &self.hsid,
991
            rend_pt.as_inner(),
992
        );
993

            
994
        let handle_proto_error = |error| FAE::RendezvousEstablish {
995
            error,
996
            rend_pt: rend_pt.clone(),
997
        };
998
        let handler = Handler {
999
            rend_established_tx,
            rend2_tx,
        };
        rend_circ
            .m_start_conversation_last_hop(Some(message.into()), handler)
            .await
            .map_err(handle_proto_error)?;
        // `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(handle_proto_error).await?;
        debug!(
            "hs conn to {}: RPT {}: got RENDEZVOUS_ESTABLISHED",
            &self.hsid,
            rend_pt.as_inner(),
        );
        Ok(Rendezvous {
            rend_circ,
            rend_cookie,
            rend_relay,
            rend2_rx,
            marker: PhantomData,
        })
    }
    /// Attempt (once) to send an INTRODUCE1 and wait for the INTRODUCE_ACK
    ///
    /// `take`s the input `rednezvous` (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.
    async fn exchange_introduce(
        &'c self,
        ipt: &UsableIntroPt<'_>,
        rendezvous: &mut Option<Rendezvous<'c, R, M>>,
    ) -> Result<(Rendezvous<R, M>, Introduced<R, M>), FAE> {
        let intro_index = ipt.intro_index;
        debug!(
            "hs conn to {}: IPT {}: obtaining intro circuit",
            &self.hsid, intro_index,
        );
        let intro_circ = self
            .circpool
            .m_get_or_launch_specific(
                &self.netdir,
                HsCircKind::ClientIntro,
                ipt.intro_target.clone(), // &OwnedCircTarget isn't CircTarget apparently
            )
            .await
            .map_err(|error| FAE::IntroductionCircuitObtain { error, intro_index })?;
        let rendezvous = rendezvous.take().ok_or_else(|| internal!("no rend"))?;
        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);
            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 {
            fn handle_msg(
                &mut self,
                _conversation: ConversationInHandler<'_, '_, '_>,
                msg: AnyRelayMsg,
            ) -> Result<MetaCellDisposition, tor_proto::Error> {
                self.intro_ack_tx
                    .deliver_expected_message(msg, MetaCellDisposition::ConversationFinished)
            }
        }
        let handle_intro_proto_error = |error| FAE::IntroductionExchange { error, intro_index };
        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,
        );
        intro_circ
            .m_start_conversation_last_hop(Some(intro1_real.into()), handler)
            .await
            .map_err(handle_intro_proto_error)?;
        // 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(handle_intro_proto_error)
            .await?
            .success()
            .map_err(|status| FAE::IntroductionFailed {
                status,
                intro_index,
            })?;
        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
    async fn complete_rendezvous(
        &'c self,
        ipt: &UsableIntroPt<'_>,
        rendezvous: Rendezvous<'c, R, M>,
        introduced: Introduced<R, M>,
    ) -> Result<Arc<ClientCirc!(R, M)>, FAE> {
        use tor_proto::circuit::handshake;
        let rend_pt = rend_pt_identity_for_error(&rendezvous.rend_relay);
        let intro_index = ipt.intro_index;
        let handle_proto_error = |error| FAE::RendezvousCompletionCircuitError {
            error,
            intro_index,
            rend_pt: rend_pt.clone(),
        };
        debug!(
            "hs conn to {}: RPT {} IPT {}: awaiting rendezvous completion",
            &self.hsid,
            rend_pt.as_inner(),
            intro_index,
        );
        let rend2_msg: Rendezvous2 = rendezvous.rend2_rx.recv(handle_proto_error).await?;
        debug!(
            "hs conn to {}: RPT {} IPT {}: received RENDEZVOUS2",
            &self.hsid,
            rend_pt.as_inner(),
            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
        let handshake_state = introduced.handshake_state;
        // Try to complete the cryptographic handshake.
        let keygen = handshake_state
            .client_receive_rend(rend2_msg.handshake_info())
            // 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.
            .map_err(|error| FAE::RendezvousCompletionHandshake {
                error,
                intro_index,
                rend_pt: rend_pt.clone(),
            })?;
        let params = circparameters_from_netparameters(self.netdir.params());
        rendezvous
            .rend_circ
            .m_extend_virtual(
                handshake::RelayProtocol::HsV3,
                handshake::HandshakeRole::Initiator,
                keygen,
                params,
            )
            .await
            .map_err(into_internal!(
                "actually this is probably a 'circuit closed' error" // TODO HS
            ))?;
        debug!(
            "hs conn to {}: RPT {} IPT {}: HS circuit established",
            &self.hsid,
            rend_pt.as_inner(),
            intro_index,
        );
        Ok(rendezvous.rend_circ)
    }
    /// Helper to estimate a timeout for a complicated operation
    ///
    /// `actions` is a list of `(count, action)`, where each entry
    /// represents doing `action`, `count` times sequentially.
    ///
    /// Combines the timeout estimates and returns an overall timeout.
    fn estimate_timeout(&self, actions: &[(u32, TimeoutsAction)]) -> Duration {
        // This algorithm is, perhaps, wrong.  For uncorrelated variables, a particular
        // percentile estimate for a sum of random variables, is not calculated by adding the
        // percentile estimates of the individual variables.
        //
        // But the actual lengths of times of the operations aren't uncorrelated.
        // If they were *perfectly* correlated, then this addition would be correct.
        // It will do for now; it just might be rather longer than it ought to be.
        actions
            .iter()
            .map(|(count, action)| {
                self.circpool
                    .m_estimate_timeout(action)
                    .saturating_mul(*count)
            })
            .fold(Duration::ZERO, Duration::saturating_add)
    }
}
/// Mocks used for testing `connect.rs`
///
/// This is different to `MockableConnectorData`,
/// which is used to *replace* this file, when testing `state.rs`.
///
/// `MocksForConnect` provides mock facilities for *testing* this file.
//
// TODO this should probably live somewhere else, maybe tor-circmgr even?
// TODO this really ought to be made by macros or something
trait MocksForConnect<R>: Clone {
    /// HS circuit pool
    type HsCircPool: MockableCircPool<R>;
    /// A random number generator
    type Rng: rand::Rng + rand::CryptoRng;
    /// Tell tests we got this descriptor text
    fn test_got_desc(&self, _: &HsDesc) {}
    /// Tell tests we got this circuit
    fn test_got_circ(&self, _: &Arc<ClientCirc!(R, Self)>) {}
    /// Tell tests we have obtained and sorted the intros like this
    fn test_got_ipts(&self, _: &[UsableIntroPt]) {}
    /// Return a random number generator
    fn thread_rng(&self) -> Self::Rng;
}
/// Mock for `HsCircPool`
///
/// Methods start with `m_` to avoid the following problem:
/// `ClientCirc::start_conversation` (say) means
/// to use the inherent method if one exists,
/// but will use a trait method if there isn't an inherent method.
///
/// So if the inherent method is renamed, the call in the impl here
/// turns into an always-recursive call.
/// This is not detected by the compiler due to the situation being
/// complicated by futures, `#[async_trait]` etc.
/// <https://github.com/rust-lang/rust/issues/111177>
#[async_trait]
trait MockableCircPool<R> {
    /// Client circuit
    type ClientCirc: MockableClientCirc;
    async fn m_get_or_launch_specific(
        &self,
        netdir: &NetDir,
        kind: HsCircKind,
        target: impl CircTarget + Send + Sync + 'async_trait,
    ) -> tor_circmgr::Result<Arc<Self::ClientCirc>>;
    /// Client circuit
    async fn m_get_or_launch_client_rend<'a>(
        &self,
        netdir: &'a NetDir,
    ) -> tor_circmgr::Result<(Arc<Self::ClientCirc>, Relay<'a>)>;
    /// Estimate timeout
    fn m_estimate_timeout(&self, action: &TimeoutsAction) -> Duration;
}
/// Mock for `ClientCirc`
#[async_trait]
trait MockableClientCirc: Debug {
    /// Client circuit
    type DirStream: AsyncRead + AsyncWrite + Send + Unpin;
    async fn m_begin_dir_stream(self: Arc<Self>) -> tor_proto::Result<Self::DirStream>;
    /// Converse
    async fn m_start_conversation_last_hop(
        &self,
        msg: Option<AnyRelayMsg>,
        reply_handler: impl MsgHandler + Send + 'static,
    ) -> tor_proto::Result<Self::Conversation<'_>>;
    /// Conversation
    type Conversation<'r>
    where
        Self: 'r;
    /// Add a virtual hop to the circuit.
    async fn m_extend_virtual(
        &self,
        protocol: tor_proto::circuit::handshake::RelayProtocol,
        role: tor_proto::circuit::handshake::HandshakeRole,
        handshake: impl tor_proto::circuit::handshake::KeyGenerator + Send,
        params: CircParameters,
    ) -> tor_proto::Result<()>;
}
impl<R: Runtime> MocksForConnect<R> for () {
    type HsCircPool = HsCircPool<R>;
    type Rng = rand::rngs::ThreadRng;
    fn thread_rng(&self) -> Self::Rng {
        rand::thread_rng()
    }
}
#[async_trait]
impl<R: Runtime> MockableCircPool<R> for HsCircPool<R> {
    type ClientCirc = ClientCirc;
    async fn m_get_or_launch_specific(
        &self,
        netdir: &NetDir,
        kind: HsCircKind,
        target: impl CircTarget + Send + Sync + 'async_trait,
    ) -> tor_circmgr::Result<Arc<ClientCirc>> {
        HsCircPool::get_or_launch_specific(self, netdir, kind, target).await
    }
    async fn m_get_or_launch_client_rend<'a>(
        &self,
        netdir: &'a NetDir,
    ) -> tor_circmgr::Result<(Arc<ClientCirc>, Relay<'a>)> {
        HsCircPool::get_or_launch_client_rend(self, netdir).await
    }
    fn m_estimate_timeout(&self, action: &TimeoutsAction) -> Duration {
        HsCircPool::estimate_timeout(self, action)
    }
}
#[async_trait]
impl MockableClientCirc for ClientCirc {
    /// Client circuit
    type DirStream = tor_proto::stream::DataStream;
    async fn m_begin_dir_stream(self: Arc<Self>) -> tor_proto::Result<Self::DirStream> {
        ClientCirc::begin_dir_stream(self).await
    }
    async fn m_start_conversation_last_hop(
        &self,
        msg: Option<AnyRelayMsg>,
        reply_handler: impl MsgHandler + Send + 'static,
    ) -> tor_proto::Result<Self::Conversation<'_>> {
        let last_hop = self.last_hop_num()?;
        ClientCirc::start_conversation(self, msg, reply_handler, last_hop).await
    }
    type Conversation<'r> = tor_proto::circuit::Conversation<'r>;
    async fn m_extend_virtual(
        &self,
        protocol: tor_proto::circuit::handshake::RelayProtocol,
        role: tor_proto::circuit::handshake::HandshakeRole,
        handshake: impl tor_proto::circuit::handshake::KeyGenerator + Send,
        params: CircParameters,
    ) -> tor_proto::Result<()> {
        ClientCirc::extend_virtual(self, protocol, role, handshake, params).await
    }
}
#[async_trait]
impl MockableConnectorData for Data {
    type ClientCirc = ClientCirc;
    type MockGlobalState = ();
    async fn connect<R: Runtime>(
        connector: &HsClientConnector<R>,
        netdir: Arc<NetDir>,
        config: Arc<Config>,
        hsid: HsId,
        data: &mut Self,
        secret_keys: HsClientSecretKeys,
    ) -> Result<Arc<Self::ClientCirc>, ConnError> {
        connect(connector, netdir, config, hsid, data, secret_keys).await
    }
    fn circuit_is_ok(circuit: &Self::ClientCirc) -> bool {
        !circuit.is_closing()
    }
}
#[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::print_stderr)]
    #![allow(clippy::print_stdout)]
    #![allow(clippy::single_char_pattern)]
    #![allow(clippy::unwrap_used)]
    #![allow(clippy::unchecked_duration_subtraction)]
    #![allow(clippy::useless_vec)]
    #![allow(clippy::needless_pass_by_value)]
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
    #![allow(dead_code, unused_variables)] // TODO HS TESTS delete, after tests are completed
    use super::*;
    use crate::*;
    use futures::FutureExt as _;
    use std::ops::{Bound, RangeBounds};
    use std::{iter, panic::AssertUnwindSafe};
    use tokio_crate as tokio;
    use tor_async_utils::JoinReadWrite;
    use tor_basic_utils::test_rng::{testing_rng, TestingRng};
    use tor_hscrypto::pk::{HsClientDescEncKey, HsClientDescEncKeypair};
    use tor_llcrypto::pk::curve25519;
    use tor_netdoc::doc::{hsdesc::test_data, netstatus::Lifetime};
    use tor_rtcompat::{tokio::TokioNativeTlsRuntime, CompoundRuntime};
    use tor_rtmock::time::MockSleepProvider;
    use tracing_test::traced_test;
    #[derive(Debug, Default)]
    struct MocksGlobal {
        hsdirs_asked: Vec<OwnedCircTarget>,
        got_desc: Option<HsDesc>,
    }
    #[derive(Clone, Debug)]
    struct Mocks<I> {
        mglobal: Arc<Mutex<MocksGlobal>>,
        id: I,
    }
    impl<I> Mocks<I> {
        fn map_id<J>(&self, f: impl FnOnce(&I) -> J) -> Mocks<J> {
            Mocks {
                mglobal: self.mglobal.clone(),
                id: f(&self.id),
            }
        }
    }
    impl<R: Runtime> MocksForConnect<R> for Mocks<()> {