1
//! Implement a cache for onion descriptors and the facility to remember a bit
2
//! about onion service history.
3

            
4
use std::fmt::Debug;
5
use std::mem;
6
use std::panic::AssertUnwindSafe;
7
use std::sync::{Arc, Mutex, MutexGuard};
8
use std::time::{Duration, Instant};
9

            
10
use futures::FutureExt as _;
11
use futures::task::{SpawnError, SpawnExt as _};
12

            
13
use async_trait::async_trait;
14
use educe::Educe;
15
use either::Either::{self, *};
16
use postage::stream::Stream as _;
17
use tracing::{debug, error, trace};
18

            
19
use safelog::DisplayRedacted as _;
20
use tor_basic_utils::define_accessor_trait;
21
use tor_circmgr::isolation::Isolation;
22
use tor_error::{Bug, ErrorReport as _, debug_report, error_report, internal};
23
use tor_hscrypto::pk::HsId;
24
use tor_netdir::NetDir;
25
use tor_rtcompat::Runtime;
26

            
27
use crate::isol_map;
28
use crate::{ConnError, HsClientConnector, HsClientSecretKeys};
29

            
30
slotmap_careful::new_key_type! {
31
    struct TableIndex;
32
}
33

            
34
/// Configuration, currently just some retry parameters
35
#[derive(Default, Debug)]
36
// This is not really public.
37
// It has to be `pub` because it appears in one of the methods in `MockableConnectorData`.
38
// That has to be because that trait is a bound on a parameter for `HsClientConnector`.
39
// `Config` is not re-exported.  (This is isomorphic to the trait sealing pattern.)
40
//
41
// This means that this struct cannot live in the crate root, so we put it here.
42
pub struct Config {
43
    /// Retry parameters
44
    pub(crate) retry: tor_circmgr::CircuitTiming,
45
}
46

            
47
define_accessor_trait! {
48
    /// Configuration for an HS client connector
49
    ///
50
    /// If the HS client connector gains new configurabilities, this trait will gain additional
51
    /// supertraits, as an API break.
52
    ///
53
    /// Prefer to use `TorClientConfig`, which will always implement this trait.
54
    //
55
    // This arrangement is very like that for `CircMgrConfig`.
56
    pub trait HsClientConnectorConfig {
57
        circuit_timing: tor_circmgr::CircuitTiming,
58
    }
59
}
60

            
61
/// Number of times we're willing to iterate round the state machine loop
62
///
63
/// **Not** the number of retries of failed descriptor downloads, circuits, etc.
64
///
65
/// The state machine loop is a condition variable loop.
66
/// It repeatedly transforms the [`ServiceState`] to try to get to `Open`,
67
/// converting stale data to `Closed` and `Closed` to `Working`, and so on.
68
/// This ought only to go forwards so in principle we could use an infinite loop.
69
/// But if we have a logic error, we want to crash eventually.
70
/// The `rechecks` counter is for detecting such a situation.
71
///
72
/// This is fairly arbitrary, but we shouldn't get anywhere near it.
73
///
74
/// Note that this is **not** a number of operational retries
75
/// of fallible retriable operations.
76
/// Such retries are handled in [`connect.rs`](crate::connect).
77
const MAX_RECHECKS: u32 = 10;
78

            
79
/// C Tor `MaxCircuitDirtiness`
80
///
81
/// As per
82
///    <https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914433>
83
///
84
/// And C Tor's `tor(1)`, which says:
85
///
86
/// > MaxCircuitDirtiness NUM
87
/// >
88
/// > Feel free to reuse a circuit that was first used at most NUM
89
/// > seconds ago, but never attach a new stream to a circuit that is
90
/// > too old.  For hidden services, this applies to the last time a
91
/// > circuit was used, not the first.  Circuits with streams
92
/// > constructed with SOCKS authentication via SocksPorts that have
93
/// > KeepAliveIsolateSOCKSAuth also remain alive for
94
/// > MaxCircuitDirtiness seconds after carrying the last such
95
/// > stream. (Default: 10 minutes)
96
///
97
/// However, we're not entirely sure this is the right behaviour.
98
/// See <https://gitlab.torproject.org/tpo/core/arti/-/issues/916>
99
///
100
// TODO SPEC: Explain C Tor `MaxCircuitDirtiness` behaviour
101
//
102
// TODO HS CFG: This should be configurable somehow
103
const RETAIN_CIRCUIT_AFTER_LAST_USE: Duration = Duration::from_secs(10 * 60);
104

            
105
/// How long to retain cached data about a hidden service
106
///
107
/// This is simply to reclaim space, not for correctness.
108
/// So we only check this during housekeeping, not operation.
109
///
110
/// The starting point for this interval is the last time we used the data,
111
/// or a circuit derived from it.
112
///
113
/// Note that this is a *maximum* for the length of time we will retain a descriptor;
114
/// HS descriptors' lifetimes (as declared in the descriptor) *are* honoured;
115
/// but that's done by the code in `connect.rs`, not here.
116
///
117
/// We're not sure this is the right value.
118
/// See <https://gitlab.torproject.org/tpo/core/arti/-/issues/916>
119
//
120
// TODO SPEC: State how long IPT and descriptor data should be retained after use
121
//
122
// TODO HS CFG: Perhaps this should be configurable somehow?
123
const RETAIN_DATA_AFTER_LAST_USE: Duration = Duration::from_secs(48 * 3600 /*hours*/);
124

            
125
/// Hidden services;, our connections to them, and history of connections, etc.
126
///
127
/// Table containing state of our ideas about services.
128
/// Data structure is keyed (indexed) by:
129
///  * `HsId`, hidden service identity
130
///  * any secret keys we are to use
131
///  * circuit isolation
132
///
133
/// We treat different values for any of the above as completely independent,
134
/// except that we try isolation joining (narrowing) if everything else matches.
135
///
136
/// In other words,
137
///  * Two HS connection requests cannot share state and effort
138
///    (descriptor downloads, descriptors, intro pt history)
139
///    unless the restricted discovery keys to be used are the same.
140
///  * This criterion is checked before looking at isolations,
141
///    which may further restrict sharing:
142
///    Two HS connection requests will only share state subject to isolations.
143
///
144
/// Here "state and effort" includes underlying circuits such as hsdir circuits,
145
/// since each HS connection state will use `launch_specific_isolated` for those.
146
#[derive(Default, Debug)]
147
pub(crate) struct Services<D: MockableConnectorData> {
148
    /// The actual records of our connections/attempts for each service, as separated
149
    records: isol_map::MultikeyIsolatedMap<TableIndex, HsId, HsClientSecretKeys, ServiceState<D>>,
150

            
151
    /// Configuration
152
    ///
153
    /// `Arc` so that it can be shared with individual hs connector tasks
154
    config: Arc<Config>,
155
}
156

            
157
/// Entry in the 2nd-level lookup array
158
#[allow(dead_code)] // This alias is here for documentation if nothing else
159
type ServiceRecord<D> = isol_map::Record<HsClientSecretKeys, ServiceState<D>>;
160

            
161
/// Value in the `Services` data structure
162
///
163
/// State and history of of our connections, including connection to any connection task.
164
///
165
/// `last_used` is used to expire data eventually.
166
//
167
// TODO unify this with channels and circuits.  See arti#778.
168
#[derive(Educe)]
169
#[educe(Debug)]
170
enum ServiceState<D: MockableConnectorData> {
171
    /// We don't have a circuit
172
    Closed {
173
        /// The state
174
        data: D,
175
        /// Last time we touched this, including reuse
176
        last_used: Instant,
177
    },
178
    /// We have an open circuit, which we can (hopefully) just use
179
    Open {
180
        /// The state
181
        data: D,
182
        /// The circuit
183
        #[educe(Debug(ignore))]
184
        tunnel: Arc<D::DataTunnel>,
185
        /// Last time we touched this, including reuse
186
        ///
187
        /// This is set when we created the circuit, and updated when we
188
        /// hand out this circuit again in response to a new request.
189
        ///
190
        /// We believe this mirrors C Tor behaviour;
191
        /// see [`RETAIN_CIRCUIT_AFTER_LAST_USE`].
192
        last_used: Instant,
193
        /// We have a task that will close the circuit when required
194
        ///
195
        /// This field serves to require construction sites of Open
196
        /// to demonstrate that there *is* an expiry task.
197
        /// In the future, it may also serve to cancel old expiry tasks.
198
        circuit_expiry_task: CircuitExpiryTask,
199
    },
200
    /// We have a task trying to find the service and establish the circuit
201
    ///
202
    /// CachedData is owned by the task.
203
    Working {
204
        /// Signals instances of `get_or_launch_connection` when the task completes
205
        barrier_recv: postage::barrier::Receiver,
206
        /// Where the task will store the error.
207
        ///
208
        /// Lock hierarchy: this lock is "inside" the big lock on `Services`.
209
        error: Arc<Mutex<Option<ConnError>>>,
210
    },
211
    /// Dummy value for use with temporary mem replace
212
    Dummy,
213
}
214

            
215
impl<D: MockableConnectorData> ServiceState<D> {
216
    /// Make a new (blank) `ServiceState::Closed`
217
18
    fn blank(runtime: &impl Runtime) -> Self {
218
18
        ServiceState::Closed {
219
18
            data: D::default(),
220
18
            last_used: runtime.now(),
221
18
        }
222
18
    }
223
}
224

            
225
/// "Continuation" return type from `obtain_circuit_or_continuation_info`
226
type Continuation = (Arc<Mutex<Option<ConnError>>>, postage::barrier::Receiver);
227

            
228
/// Represents a task which is waiting to see when the circuit needs to be expired
229
///
230
/// TODO: Replace this with a task handle that cancels the task when dropped.
231
/// Until then, if the circuit is closed before then, the expiry task will
232
/// uselessly wake up some time later.
233
#[derive(Debug)] // Not Clone
234
struct CircuitExpiryTask {}
235
// impl Drop already, partly to allow explicit drop(CircuitExpiryTask) without clippy complaint
236
impl Drop for CircuitExpiryTask {
237
12
    fn drop(&mut self) {}
238
}
239

            
240
/// Obtain a circuit from the `Services` table, or return a continuation
241
///
242
/// This is the workhorse function for `get_or_launch_connection`.
243
///
244
/// `get_or_launch_connection`, together with `obtain_circuit_or_continuation_info`,
245
/// form a condition variable loop:
246
///
247
/// We check to see if we have a circuit.  If so, we return it.
248
/// Otherwise, we make sure that a circuit is being constructed,
249
/// and then go into a condvar wait;
250
/// we'll be signaled when the construction completes.
251
///
252
/// So the connection task we spawn does not return the circuit, or error,
253
/// via an inter-task stream.
254
/// It stores it in the data structure and wakes up all the client tasks.
255
/// (This means there is only one success path for the client task code.)
256
///
257
/// There are some wrinkles:
258
///
259
/// ### Existence of this as a separate function
260
///
261
/// The usual structure for a condition variable loop would be something like this:
262
///
263
/// ```rust,ignore
264
/// loop {
265
///    test state and maybe break;
266
///    cv.wait(guard).await; // consumes guard, unlocking after enqueueing us as a waiter
267
///    guard = lock();
268
/// }
269
/// ```
270
///
271
/// However, Rust does not currently understand that the mutex is not
272
/// actually a captured variable held across an await point,
273
/// when the variable is consumed before the await, and re-stored afterwards.
274
/// As a result, the async future becomes erroneously `!Send`:
275
/// <https://github.com/rust-lang/rust/issues/104883>.
276
/// We want the unstable feature `-Zdrop-tracking`:
277
/// <https://github.com/rust-lang/rust/issues/97331>.
278
///
279
/// Instead, to convince the compiler, we must use a scope-based drop of the mutex guard.
280
/// That means converting the "test state and maybe break" part into a sub-function.
281
/// That's what this function is.
282
///
283
/// It returns `Right` if the loop should be exited, returning the circuit to the caller.
284
/// It returns `Left` if the loop needs to do a condition variable wait.
285
///
286
/// ### We're using a barrier as a condition variable
287
///
288
/// We want to be signaled when the task exits.  Indeed, *only* when it exits.
289
/// This functionality is most conveniently in a `postage::barrier`.
290
///
291
/// ### Nested loops
292
///
293
/// Sometimes we want to go round again *without* unlocking.
294
/// Sometimes we must unlock and wait and relock.
295
///
296
/// The drop tracking workaround (see above) means we have to do these two
297
/// in separate scopes.
298
/// So there are two nested loops: one here, and one in `get_or_launch_connection`.
299
/// They both use the same backstop rechecks counter.
300
68
fn obtain_circuit_or_continuation_info<D: MockableConnectorData>(
301
68
    connector: &HsClientConnector<impl Runtime, D>,
302
68
    netdir: &Arc<NetDir>,
303
68
    hsid: &HsId,
304
68
    secret_keys: &HsClientSecretKeys,
305
68
    table_index: TableIndex,
306
68
    rechecks: &mut impl Iterator,
307
68
    mut guard: MutexGuard<'_, Services<D>>,
308
68
) -> Result<Either<Continuation, Arc<D::DataTunnel>>, ConnError> {
309
68
    let blank_state = || ServiceState::blank(&connector.runtime);
310

            
311
94
    for _recheck in rechecks {
312
94
        let record = guard
313
94
            .records
314
94
            .by_index_mut(table_index)
315
94
            .ok_or_else(|| internal!("guard table entry vanished!"))?;
316
94
        let state = &mut **record;
317

            
318
94
        trace!("HS conn state: {state:?}");
319

            
320
94
        let (data, barrier_send) = match state {
321
            ServiceState::Open {
322
                data: _,
323
40
                tunnel,
324
40
                last_used,
325
                circuit_expiry_task: _,
326
            } => {
327
40
                let now = connector.runtime.now();
328
40
                if !D::tunnel_is_ok(tunnel) {
329
                    // Well that's no good, we need a fresh one, but keep the data
330
                    let data = match mem::replace(state, ServiceState::Dummy) {
331
                        ServiceState::Open {
332
                            data,
333
                            last_used: _,
334
                            tunnel: _,
335
                            circuit_expiry_task: _,
336
                        } => data,
337
                        _ => panic!("state changed between matches"),
338
                    };
339
                    *state = ServiceState::Closed {
340
                        data,
341
                        last_used: now,
342
                    };
343
                    continue;
344
40
                }
345
40
                *last_used = now;
346
                // No need to tell expiry task about revised expiry time;
347
                // it will see the new last_used when it wakes up at the old expiry time.
348

            
349
40
                return Ok::<_, ConnError>(Right(tunnel.clone()));
350
            }
351
            ServiceState::Working {
352
28
                barrier_recv,
353
28
                error,
354
            } => {
355
                if !matches!(
356
28
                    barrier_recv.try_recv(),
357
                    Err(postage::stream::TryRecvError::Pending)
358
                ) {
359
                    // This information is stale; the task no longer exists.
360
                    // We want information from a fresh task.
361
                    *state = blank_state();
362
                    continue;
363
28
                }
364
28
                let barrier_recv = barrier_recv.clone();
365

            
366
                // This clone of the error field Arc<Mutex<..>> allows us to collect errors
367
                // which happened due to the currently-running task, which we have just
368
                // found exists.  Ie, it will see errors that occurred after we entered
369
                // `get_or_launch`.  Stale errors, from previous tasks, were cleared above.
370
28
                let error = error.clone();
371

            
372
                // Wait for the task to complete (at which point it drops the barrier)
373
28
                return Ok(Left((error, barrier_recv)));
374
            }
375
            ServiceState::Closed { .. } => {
376
26
                let (barrier_send, barrier_recv) = postage::barrier::channel();
377
26
                let data = match mem::replace(
378
26
                    state,
379
26
                    ServiceState::Working {
380
26
                        barrier_recv,
381
26
                        error: Arc::new(Mutex::new(None)),
382
26
                    },
383
26
                ) {
384
26
                    ServiceState::Closed { data, .. } => data,
385
                    _ => panic!("state changed between matches"),
386
                };
387
26
                (data, barrier_send)
388
            }
389
            ServiceState::Dummy => {
390
                *state = blank_state();
391
                return Err(internal!("HS connector found dummy state").into());
392
            }
393
        };
394

            
395
        // Make a connection
396
26
        let runtime = &connector.runtime;
397
26
        let connector = (*connector).clone();
398
26
        let config = guard.config.clone();
399
26
        let netdir = netdir.clone();
400
26
        let secret_keys = secret_keys.clone();
401
26
        let hsid = *hsid;
402
26
        let connect_future = async move {
403
26
            let mut data = data;
404

            
405
26
            let got = AssertUnwindSafe(D::connect(
406
26
                &connector,
407
26
                netdir,
408
26
                config,
409
26
                hsid,
410
26
                &mut data,
411
26
                secret_keys,
412
26
            ))
413
26
            .catch_unwind()
414
26
            .await
415
26
            .unwrap_or_else(|_| {
416
                data = D::default();
417
                Err(internal!("hidden service connector task panicked!").into())
418
            });
419
26
            let now = connector.runtime.now();
420
26
            let last_used = now;
421

            
422
26
            let got = got.and_then(|circuit| {
423
26
                let circuit_expiry_task = ServiceState::spawn_circuit_expiry_task(
424
26
                    &connector,
425
26
                    hsid,
426
26
                    table_index,
427
26
                    last_used,
428
26
                    now,
429
                )
430
26
                .map_err(|cause| ConnError::Spawn {
431
                    spawning: "circuit expiry task",
432
                    cause: cause.into(),
433
                })?;
434
26
                Ok((circuit, circuit_expiry_task))
435
26
            });
436

            
437
26
            let got_error = got.as_ref().map(|_| ()).map_err(Clone::clone);
438

            
439
            // block for handling inability to store
440
26
            let stored = async {
441
26
                let mut guard = connector.services()?;
442
26
                let record = guard
443
26
                    .records
444
26
                    .by_index_mut(table_index)
445
26
                    .ok_or_else(|| internal!("HS table entry removed while task running"))?;
446
                // Always match this, so we check what we're overwriting
447
26
                let state = &mut **record;
448
26
                let error_store = match state {
449
26
                    ServiceState::Working { error, .. } => error,
450
                    _ => return Err(internal!("HS task found state other than Working")),
451
                };
452

            
453
26
                match got {
454
26
                    Ok((tunnel, circuit_expiry_task)) => {
455
26
                        *state = ServiceState::Open {
456
26
                            data,
457
26
                            tunnel: Arc::new(tunnel),
458
26
                            last_used,
459
26
                            circuit_expiry_task,
460
26
                        }
461
                    }
462
                    Err(error) => {
463
                        let mut error_store = error_store
464
                            .lock()
465
                            .map_err(|_| internal!("Working error poisoned, cannot store error"))?;
466
                        *error_store = Some(error);
467
                    }
468
                };
469

            
470
26
                Ok(())
471
26
            }
472
26
            .await;
473

            
474
26
            match (got_error, stored) {
475
26
                (Ok::<(), ConnError>(()), Ok::<(), Bug>(())) => {}
476
                (Err(got_error), Ok(())) => {
477
                    debug_report!(
478
                        got_error,
479
                        "HS connection failure for {}",
480
                        hsid.display_redacted()
481
                    );
482
                }
483
                (Ok(()), Err(bug)) => {
484
                    error_report!(
485
                        bug,
486
                        "internal error storing built HS circuit for {}",
487
                        hsid.display_redacted()
488
                    );
489
                }
490
                (Err(got_error), Err(bug)) => {
491
                    // We're reporting two errors, so we'll construct the event
492
                    // manually.
493
                    error!(
494
                        "internal error storing HS connection error for {}: {}; {}",
495
                        hsid.display_redacted(),
496
                        got_error.report(),
497
                        bug.report(),
498
                    );
499
                }
500
            };
501
26
            drop(barrier_send);
502
26
        };
503
26
        runtime
504
26
            .spawn_obj(Box::new(connect_future).into())
505
26
            .map_err(|cause| ConnError::Spawn {
506
                spawning: "connection task",
507
                cause: cause.into(),
508
            })?;
509
    }
510

            
511
    Err(internal!("HS connector state management malfunction (exceeded MAX_RECHECKS").into())
512
68
}
513

            
514
impl<D: MockableConnectorData> Services<D> {
515
    /// Create a new empty `Services`
516
14
    pub(crate) fn new(config: Config) -> Self {
517
14
        Services {
518
14
            records: Default::default(),
519
14
            config: Arc::new(config),
520
14
        }
521
14
    }
522

            
523
    /// Connect to a hidden service
524
    // We *do* drop guard.  There is *one* await point, just after drop(guard).
525
40
    pub(crate) async fn get_or_launch_connection(
526
40
        connector: &HsClientConnector<impl Runtime, D>,
527
40
        netdir: &Arc<NetDir>,
528
40
        hs_id: HsId,
529
40
        isolation: Box<dyn Isolation>,
530
40
        secret_keys: HsClientSecretKeys,
531
40
    ) -> Result<Arc<D::DataTunnel>, ConnError> {
532
40
        let blank_state = || ServiceState::blank(&connector.runtime);
533

            
534
40
        let mut rechecks = 0..MAX_RECHECKS;
535

            
536
68
        let mut obtain = |table_index, guard| {
537
68
            obtain_circuit_or_continuation_info(
538
68
                connector,
539
68
                netdir,
540
68
                &hs_id,
541
68
                &secret_keys,
542
68
                table_index,
543
68
                &mut rechecks,
544
68
                guard,
545
            )
546
68
        };
547

            
548
        let mut got;
549
        let table_index;
550
        {
551
40
            let mut guard = connector.services()?;
552
40
            let services = &mut *guard;
553

            
554
40
            trace!("HS conn get_or_launch: {hs_id:?} {isolation:?} {secret_keys:?}");
555
            //trace!("HS conn services: {services:?}");
556

            
557
40
            table_index =
558
40
                services
559
40
                    .records
560
40
                    .index_or_insert_with(&hs_id, &secret_keys, isolation, blank_state);
561

            
562
40
            let guard = guard;
563
40
            got = obtain(table_index, guard);
564
        }
565
        loop {
566
            // The parts of this loop which run after a `Left` is returned
567
            // logically belong in the case in `obtain_circuit_or_continuation_info`
568
            // for `ServiceState::Working`, where that function decides we need to wait.
569
            // This code has to be out here to help the compiler's drop tracking.
570
            {
571
                // Block to scope the acquisition of `error`, a guard
572
                // for the mutex-protected error field in the state,
573
                // and, for neatness, barrier_recv.
574

            
575
68
                let (error, mut barrier_recv) = match got? {
576
40
                    Right(ret) => return Ok(ret),
577
28
                    Left(continuation) => continuation,
578
                };
579

            
580
28
                barrier_recv.recv().await;
581

            
582
28
                let error = error
583
28
                    .lock()
584
28
                    .map_err(|_| internal!("Working error poisoned"))?;
585
28
                if let Some(error) = &*error {
586
                    return Err(error.clone());
587
28
                }
588
            }
589

            
590
28
            let guard = connector.services()?;
591

            
592
28
            got = obtain(table_index, guard);
593
        }
594
40
    }
595

            
596
    /// Perform housekeeping - delete data we aren't interested in any more
597
20
    pub(crate) fn run_housekeeping(&mut self, now: Instant) {
598
20
        self.expire_old_data(now);
599
20
    }
600

            
601
    /// Delete data we aren't interested in any more
602
20
    fn expire_old_data(&mut self, now: Instant) {
603
20
        self.records
604
20
            .retain(|hsid, record, _table_index| match &**record {
605
12
                ServiceState::Closed { data: _, last_used } => {
606
12
                    let Some(expiry_time) = last_used.checked_add(RETAIN_DATA_AFTER_LAST_USE)
607
                    else {
608
                        return false;
609
                    };
610
12
                    now <= expiry_time
611
                }
612
8
                ServiceState::Open { .. } | ServiceState::Working { .. } => true,
613
                ServiceState::Dummy => {
614
                    error!(
615
                        "bug: found dummy data during HS housekeeping, for {}",
616
                        hsid.display_redacted()
617
                    );
618
                    false
619
                }
620
20
            });
621
20
    }
622
}
623

            
624
impl<D: MockableConnectorData> ServiceState<D> {
625
    /// Spawn a task that will drop our reference to the rendezvous circuit
626
    /// at `table_index` when it has gone too long without any use.
627
    ///
628
    /// According to [`RETAIN_CIRCUIT_AFTER_LAST_USE`].
629
    //
630
    // As it happens, this function is always called with `last_used` equal to `now`,
631
    // but we pass separate arguments for clarity.
632
26
    fn spawn_circuit_expiry_task(
633
26
        connector: &HsClientConnector<impl Runtime, D>,
634
26
        hsid: HsId,
635
26
        table_index: TableIndex,
636
26
        last_used: Instant,
637
26
        now: Instant,
638
26
    ) -> Result<CircuitExpiryTask, SpawnError> {
639
        /// Returns the duration until expiry, or `None` if it should expire now
640
42
        fn calculate_expiry_wait(last_used: Instant, now: Instant) -> Option<Duration> {
641
42
            let expiry = last_used
642
42
                .checked_add(RETAIN_CIRCUIT_AFTER_LAST_USE)
643
42
                .or_else(|| {
644
                    error!("bug: time overflow calculating HS circuit expiry, killing circuit!");
645
                    None
646
                })?;
647
42
            let wait = expiry.checked_duration_since(now).unwrap_or_default();
648
42
            if wait == Duration::ZERO {
649
12
                return None;
650
30
            }
651
30
            Some(wait)
652
42
        }
653

            
654
26
        let mut maybe_wait = calculate_expiry_wait(last_used, now);
655
26
        let () = connector.runtime.spawn({
656
26
            let connector = connector.clone();
657
26
            async move {
658
                // This loop is slightly odd.  The wait ought naturally to be at the end,
659
                // but that would mean a useless re-lock and re-check right after creation,
660
                // or jumping into the middle of the loop.
661
                loop {
662
30
                    if let Some(yes_wait) = maybe_wait {
663
30
                        connector.runtime.sleep(yes_wait).await;
664
                    }
665
                    // If it's None, we can't rely on that to say we should expire it,
666
                    // since that information crossed a time when we didn't hold the lock.
667

            
668
16
                    let Ok(mut guard) = connector.services() else {
669
                        break;
670
                    };
671
16
                    let Some(record) = guard.records.by_index_mut(table_index) else {
672
                        break;
673
                    };
674
16
                    let state = &mut **record;
675
16
                    let last_used = match state {
676
                        ServiceState::Closed { .. } => break,
677
16
                        ServiceState::Open { last_used, .. } => *last_used,
678
                        ServiceState::Working { .. } => break, // someone else will respawn
679
                        ServiceState::Dummy => break,          // someone else will (report and) fix
680
                    };
681
16
                    maybe_wait = calculate_expiry_wait(last_used, connector.runtime.now());
682
16
                    if maybe_wait.is_none() {
683
12
                        match mem::replace(state, ServiceState::Dummy) {
684
                            ServiceState::Open {
685
12
                                data,
686
12
                                tunnel: circuit,
687
12
                                last_used,
688
12
                                circuit_expiry_task,
689
                            } => {
690
12
                                debug!("HS connection expires: {hsid:?}");
691
12
                                drop(circuit);
692
12
                                drop(circuit_expiry_task); // that's us
693
12
                                *state = ServiceState::Closed { data, last_used };
694
12
                                break;
695
                            }
696
                            _ => panic!("state now {state:?} even though we just saw it Open"),
697
                        }
698
4
                    }
699
                }
700
12
            }
701
        })?;
702
26
        Ok(CircuitExpiryTask {})
703
26
    }
704
}
705

            
706
/// Mocking for actual HS connection work, to let us test the `Services` state machine
707
//
708
// Does *not* mock circmgr, chanmgr, etc. - those won't be used by the tests, since our
709
// `connect` won't call them.  But mocking them pollutes many types with `R` and is
710
// generally tiresome.  So let's not.  Instead the tests can make dummy ones.
711
//
712
// This trait is actually crate-private, since it isn't re-exported, but it must
713
// be `pub` because it appears as a default for a type parameter in HsClientConnector.
714
#[async_trait]
715
pub trait MockableConnectorData: Default + Debug + Send + Sync + 'static {
716
    /// Client circuit
717
    type DataTunnel: Sync + Send + 'static;
718

            
719
    /// Mock state
720
    type MockGlobalState: Clone + Sync + Send + 'static;
721

            
722
    /// Connect
723
    async fn connect<R: Runtime>(
724
        connector: &HsClientConnector<R, Self>,
725
        netdir: Arc<NetDir>,
726
        config: Arc<Config>,
727
        hsid: HsId,
728
        data: &mut Self,
729
        secret_keys: HsClientSecretKeys,
730
    ) -> Result<Self::DataTunnel, ConnError>;
731

            
732
    /// Is circuit OK?  Ie, not `.is_closing()`.
733
    fn tunnel_is_ok(tunnel: &Self::DataTunnel) -> bool;
734
}
735

            
736
#[cfg(test)]
737
pub(crate) mod test {
738
    // @@ begin test lint list maintained by maint/add_warning @@
739
    #![allow(clippy::bool_assert_comparison)]
740
    #![allow(clippy::clone_on_copy)]
741
    #![allow(clippy::dbg_macro)]
742
    #![allow(clippy::mixed_attributes_style)]
743
    #![allow(clippy::print_stderr)]
744
    #![allow(clippy::print_stdout)]
745
    #![allow(clippy::single_char_pattern)]
746
    #![allow(clippy::unwrap_used)]
747
    #![allow(clippy::unchecked_duration_subtraction)]
748
    #![allow(clippy::useless_vec)]
749
    #![allow(clippy::needless_pass_by_value)]
750
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
751
    use super::*;
752
    use crate::*;
753
    use futures::{SinkExt, poll};
754
    use std::fmt;
755
    use std::task::Poll::{self, *};
756
    use tokio::pin;
757
    use tokio_crate as tokio;
758
    use tor_memquota::ArcMemoryQuotaTrackerExt as _;
759
    use tor_proto::memquota::ToplevelAccount;
760
    use tor_rtcompat::{SleepProvider, test_with_one_runtime};
761
    use tor_rtmock::MockRuntime;
762
    use tracing_test::traced_test;
763

            
764
    use ConnError as E;
765

            
766
    #[derive(Debug, Default)]
767
    struct MockData {
768
        connect_called: usize,
769
    }
770

            
771
    /// Type indicating what our `connect()` should return; it always makes a fresh MockCirc
772
    type MockGive = Poll<Result<(), E>>;
773

            
774
    #[derive(Debug, Clone)]
775
    struct MockGlobalState {
776
        // things will appear here when we have more sophisticated tests
777
        give: postage::watch::Receiver<MockGive>,
778
    }
779

            
780
    #[derive(Clone, Educe)]
781
    #[educe(Debug)]
782
    struct MockTunnel {
783
        #[educe(Debug(method = "debug_arc_mutex"))]
784
        ok: Arc<Mutex<bool>>,
785
        connect_called: usize,
786
    }
787

            
788
    fn debug_arc_mutex(val: &Arc<Mutex<impl Debug>>, f: &mut fmt::Formatter) -> fmt::Result {
789
        write!(f, "@{:?}", Arc::as_ptr(val))?;
790
        let guard = val.lock();
791
        let guard = guard.or_else(|g| {
792
            write!(f, ",POISON")?;
793
            Ok::<_, fmt::Error>(g.into_inner())
794
        })?;
795
        write!(f, " ")?;
796
        Debug::fmt(&*guard, f)
797
    }
798

            
799
    impl PartialEq for MockTunnel {
800
        fn eq(&self, other: &MockTunnel) -> bool {
801
            Arc::ptr_eq(&self.ok, &other.ok)
802
        }
803
    }
804

            
805
    impl MockTunnel {
806
        fn new(connect_called: usize) -> Self {
807
            let ok = Arc::new(Mutex::new(true));
808
            MockTunnel { ok, connect_called }
809
        }
810
    }
811

            
812
    #[async_trait]
813
    impl MockableConnectorData for MockData {
814
        type DataTunnel = MockTunnel;
815
        type MockGlobalState = MockGlobalState;
816

            
817
        async fn connect<R: Runtime>(
818
            connector: &HsClientConnector<R, MockData>,
819
            _netdir: Arc<NetDir>,
820
            _config: Arc<Config>,
821
            _hsid: HsId,
822
            data: &mut MockData,
823
            _secret_keys: HsClientSecretKeys,
824
        ) -> Result<Self::DataTunnel, E> {
825
            data.connect_called += 1;
826
            let make = {
827
                let connect_called = data.connect_called;
828
                move |()| MockTunnel::new(connect_called)
829
            };
830
            let mut give = connector.mock_for_state.give.clone();
831
            if let Ready(ret) = &*give.borrow() {
832
                return ret.clone().map(make);
833
            }
834
            loop {
835
                match give.recv().await.expect("EOF on mock_global_state stream") {
836
                    Pending => {}
837
                    Ready(ret) => return ret.map(make),
838
                }
839
            }
840
        }
841

            
842
        fn tunnel_is_ok(circuit: &Self::DataTunnel) -> bool {
843
            *circuit.ok.lock().unwrap()
844
        }
845
    }
846

            
847
    /// Makes a non-empty `HsClientSecretKeys`, containing (somehow) `kk`
848
    fn mk_keys(kk: u8) -> HsClientSecretKeys {
849
        let mut ss = [0_u8; 32];
850
        ss[0] = kk;
851
        let keypair = tor_llcrypto::pk::ed25519::Keypair::from_bytes(&ss);
852
        let mut b = HsClientSecretKeysBuilder::default();
853
        #[allow(deprecated)]
854
        b.ks_hsc_intro_auth(keypair.into());
855
        b.build().unwrap()
856
    }
857

            
858
    fn mk_hsconn<R: Runtime>(
859
        runtime: R,
860
    ) -> (
861
        HsClientConnector<R, MockData>,
862
        HsClientSecretKeys,
863
        postage::watch::Sender<MockGive>,
864
    ) {
865
        let chanmgr = tor_chanmgr::ChanMgr::new(
866
            runtime.clone(),
867
            &Default::default(),
868
            tor_chanmgr::Dormancy::Dormant,
869
            &Default::default(),
870
            ToplevelAccount::new_noop(),
871
            None,
872
        );
873
        let guardmgr = tor_guardmgr::GuardMgr::new(
874
            runtime.clone(),
875
            tor_persist::TestingStateMgr::new(),
876
            &tor_guardmgr::TestConfig::default(),
877
        )
878
        .unwrap();
879

            
880
        let circmgr = Arc::new(
881
            tor_circmgr::CircMgr::new(
882
                &tor_circmgr::TestConfig::default(),
883
                tor_persist::TestingStateMgr::new(),
884
                &runtime,
885
                Arc::new(chanmgr),
886
                &guardmgr,
887
            )
888
            .unwrap(),
889
        );
890
        let circpool = Arc::new(HsCircPool::new(&circmgr));
891
        let (give_send, give) = postage::watch::channel_with(Ready(Ok(())));
892
        let mock_for_state = MockGlobalState { give };
893
        #[allow(clippy::let_and_return)] // we'll probably add more in this function
894
        let hscc = HsClientConnector {
895
            runtime,
896
            circpool,
897
            services: Default::default(),
898
            mock_for_state,
899
        };
900
        let keys = HsClientSecretKeysBuilder::default().build().unwrap();
901
        (hscc, keys, give_send)
902
    }
903

            
904
    #[allow(clippy::unnecessary_wraps)]
905
    fn mk_isol(s: &str) -> Option<NarrowableIsolation> {
906
        Some(NarrowableIsolation(s.into()))
907
    }
908

            
909
    async fn launch_one(
910
        hsconn: &HsClientConnector<impl Runtime, MockData>,
911
        id: u8,
912
        secret_keys: &HsClientSecretKeys,
913
        isolation: Option<NarrowableIsolation>,
914
    ) -> Result<Arc<MockTunnel>, ConnError> {
915
        let netdir = tor_netdir::testnet::construct_netdir()
916
            .unwrap_if_sufficient()
917
            .unwrap();
918
        let netdir = Arc::new(netdir);
919

            
920
        let hs_id = {
921
            let mut hs_id = [0_u8; 32];
922
            hs_id[0] = id;
923
            hs_id.into()
924
        };
925
        #[allow(clippy::redundant_closure)] // srsly, that would be worse
926
        let isolation = isolation.unwrap_or_default().into();
927
        Services::get_or_launch_connection(hsconn, &netdir, hs_id, isolation, secret_keys.clone())
928
            .await
929
    }
930

            
931
    #[derive(Default, Debug, Clone)]
932
    // TODO move this to tor-circmgr under a test feature?
933
    pub(crate) struct NarrowableIsolation(pub(crate) String);
934
    impl tor_circmgr::isolation::IsolationHelper for NarrowableIsolation {
935
        fn compatible_same_type(&self, other: &Self) -> bool {
936
            self.join_same_type(other).is_some()
937
        }
938
        fn join_same_type(&self, other: &Self) -> Option<Self> {
939
            Some(if self.0.starts_with(&other.0) {
940
                self.clone()
941
            } else if other.0.starts_with(&self.0) {
942
                other.clone()
943
            } else {
944
                return None;
945
            })
946
        }
947
    }
948

            
949
    #[test]
950
    #[traced_test]
951
    fn simple() {
952
        test_with_one_runtime!(|runtime| async {
953
            let (hsconn, keys, _give_send) = mk_hsconn(runtime);
954

            
955
            let circuit = launch_one(&hsconn, 0, &keys, None).await.unwrap();
956
            eprintln!("{:?}", circuit);
957
        });
958
    }
959

            
960
    #[test]
961
    #[traced_test]
962
    fn expiry() {
963
        MockRuntime::test_with_various(|runtime| async move {
964
            // This is the amount by which we adjust clock advances to make sure we
965
            // hit more or less than a particular value, to avoid edge cases and
966
            // cope with real time advancing too.
967
            // This does *not* represent an actual delay to real test runs.
968
            const TIMEOUT_SLOP: Duration = Duration::from_secs(10);
969

            
970
            let (hsconn, keys, _give_send) = mk_hsconn(runtime.clone());
971

            
972
            let advance = |duration| {
973
                let hsconn = hsconn.clone();
974
                let runtime = &runtime;
975
                async move {
976
                    // let expiry task get going and choose its expiry (wakeup) time
977
                    runtime.progress_until_stalled().await;
978
                    // TODO: Make this use runtime.advance_by() when that's not very slow
979
                    runtime.mock_sleep().advance(duration);
980
                    // let expiry task run
981
                    runtime.progress_until_stalled().await;
982
                    hsconn.services().unwrap().run_housekeeping(runtime.now());
983
                }
984
            };
985

            
986
            // make circuit1
987
            let circuit1 = launch_one(&hsconn, 0, &keys, None).await.unwrap();
988

            
989
            // expire it
990
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE + TIMEOUT_SLOP).await;
991

            
992
            // make circuit2 (a)
993
            let circuit2a = launch_one(&hsconn, 0, &keys, None).await.unwrap();
994
            assert_ne!(circuit1, circuit2a);
995

            
996
            // nearly expire it, then reuse it
997
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE - TIMEOUT_SLOP).await;
998
            let circuit2b = launch_one(&hsconn, 0, &keys, None).await.unwrap();
999
            assert_eq!(circuit2a, circuit2b);
            // nearly expire it again, then reuse it
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE - TIMEOUT_SLOP).await;
            let circuit2c = launch_one(&hsconn, 0, &keys, None).await.unwrap();
            assert_eq!(circuit2a, circuit2c);
            // actually expire it
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE + TIMEOUT_SLOP).await;
            let circuit3 = launch_one(&hsconn, 0, &keys, None).await.unwrap();
            assert_ne!(circuit2c, circuit3);
            assert_eq!(circuit3.connect_called, 3);
            advance(RETAIN_DATA_AFTER_LAST_USE + Duration::from_secs(10)).await;
            let circuit4 = launch_one(&hsconn, 0, &keys, None).await.unwrap();
            assert_eq!(circuit4.connect_called, 1);
        });
    }
    #[test]
    #[traced_test]
    fn coalesce() {
        test_with_one_runtime!(|runtime| async {
            let (hsconn, keys, mut give_send) = mk_hsconn(runtime);
            give_send.send(Pending).await.unwrap();
            let c1f = launch_one(&hsconn, 0, &keys, None);
            pin!(c1f);
            for _ in 0..10 {
                assert!(poll!(&mut c1f).is_pending());
            }
            // c2f will find Working
            let c2f = launch_one(&hsconn, 0, &keys, None);
            pin!(c2f);
            for _ in 0..10 {
                assert!(poll!(&mut c1f).is_pending());
                assert!(poll!(&mut c2f).is_pending());
            }
            give_send.send(Ready(Ok(()))).await.unwrap();
            let c1 = c1f.await.unwrap();
            let c2 = c2f.await.unwrap();
            assert_eq!(c1, c2);
            // c2 will find Open
            let c3 = launch_one(&hsconn, 0, &keys, None).await.unwrap();
            assert_eq!(c1, c3);
            assert_ne!(c1, launch_one(&hsconn, 1, &keys, None).await.unwrap());
            assert_ne!(
                c1,
                launch_one(&hsconn, 0, &mk_keys(42), None).await.unwrap()
            );
            let c_isol_1 = launch_one(&hsconn, 0, &keys, mk_isol("a")).await.unwrap();
            assert_eq!(c1, c_isol_1); // We can reuse, but now we've narrowed the isol
            let c_isol_2 = launch_one(&hsconn, 0, &keys, mk_isol("b")).await.unwrap();
            assert_ne!(c1, c_isol_2);
        });
    }
}