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

            
9
use futures::FutureExt as _;
10
use futures::task::SpawnError;
11

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            
300
94
    for _recheck in rechecks {
301
94
        let record = guard
302
94
            .records
303
94
            .by_index_mut(table_index)
304
94
            .ok_or_else(|| internal!("guard table entry vanished!"))?;
305
94
        let state = &mut **record;
306

            
307
94
        trace!("HS conn state: {state:?}");
308

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

            
338
40
                return Ok::<_, ConnError>(Right(tunnel.clone()));
339
            }
340
            ServiceState::Working {
341
28
                barrier_recv,
342
28
                error,
343
            } => {
344
                if !matches!(
345
28
                    barrier_recv.try_recv(),
346
                    Err(postage::stream::TryRecvError::Pending)
347
                ) {
348
                    // This information is stale; the task no longer exists.
349
                    // We want information from a fresh task.
350
                    *state = blank_state();
351
                    continue;
352
28
                }
353
28
                let barrier_recv = barrier_recv.clone();
354

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

            
361
                // Wait for the task to complete (at which point it drops the barrier)
362
28
                return Ok(Left((error, barrier_recv)));
363
            }
364
            ServiceState::Closed { .. } => {
365
26
                let (barrier_send, barrier_recv) = postage::barrier::channel();
366
26
                let data = match mem::replace(
367
26
                    state,
368
26
                    ServiceState::Working {
369
26
                        barrier_recv,
370
26
                        error: Arc::new(Mutex::new(None)),
371
26
                    },
372
26
                ) {
373
26
                    ServiceState::Closed { data, .. } => data,
374
                    _ => panic!("state changed between matches"),
375
                };
376
26
                (data, barrier_send)
377
            }
378
            ServiceState::Dummy => {
379
                *state = blank_state();
380
                return Err(internal!("HS connector found dummy state").into());
381
            }
382
        };
383

            
384
        // Make a connection
385
26
        let runtime = &connector.runtime;
386
26
        let connector = (*connector).clone();
387
26
        let config = guard.config.clone();
388
26
        let netdir = netdir.clone();
389
26
        let secret_keys = secret_keys.clone();
390
26
        let hsid = *hsid;
391
26
        let connect_future = async move {
392
26
            let mut data = data;
393

            
394
26
            let got = AssertUnwindSafe(D::connect(
395
26
                &connector,
396
26
                netdir,
397
26
                config,
398
26
                hsid,
399
26
                &mut data,
400
26
                secret_keys,
401
26
            ))
402
26
            .catch_unwind()
403
26
            .await
404
26
            .unwrap_or_else(|_| {
405
                data = D::default();
406
                Err(internal!("hidden service connector task panicked!").into())
407
            });
408
26
            let now = connector.runtime.now();
409
26
            let last_used = now;
410

            
411
26
            let got = got.and_then(|circuit| {
412
26
                let circuit_expiry_task = ServiceState::spawn_circuit_expiry_task(
413
26
                    &connector,
414
26
                    hsid,
415
26
                    table_index,
416
26
                    last_used,
417
26
                    now,
418
                )
419
26
                .map_err(|cause| ConnError::Spawn {
420
                    spawning: "circuit expiry task",
421
                    cause: cause.into(),
422
                })?;
423
26
                Ok((circuit, circuit_expiry_task))
424
26
            });
425

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

            
428
            // block for handling inability to store
429
26
            let stored = async {
430
26
                let mut guard = connector.services()?;
431
26
                let record = guard
432
26
                    .records
433
26
                    .by_index_mut(table_index)
434
26
                    .ok_or_else(|| internal!("HS table entry removed while task running"))?;
435
                // Always match this, so we check what we're overwriting
436
26
                let state = &mut **record;
437
26
                let error_store = match state {
438
26
                    ServiceState::Working { error, .. } => error,
439
                    _ => return Err(internal!("HS task found state other than Working")),
440
                };
441

            
442
26
                match got {
443
26
                    Ok((tunnel, circuit_expiry_task)) => {
444
26
                        *state = ServiceState::Open {
445
26
                            data,
446
26
                            tunnel: Arc::new(tunnel),
447
26
                            last_used,
448
26
                            circuit_expiry_task,
449
26
                        }
450
                    }
451
                    Err(error) => {
452
                        let mut error_store = error_store
453
                            .lock()
454
                            .map_err(|_| internal!("Working error poisoned, cannot store error"))?;
455
                        *error_store = Some(error);
456
                    }
457
                };
458

            
459
26
                Ok(())
460
26
            }
461
26
            .await;
462

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

            
500
    Err(internal!("HS connector state management malfunction (exceeded MAX_RECHECKS").into())
501
68
}
502

            
503
impl<D: MockableConnectorData> Services<D> {
504
    /// Create a new empty `Services`
505
    pub(crate) fn new(config: Config) -> Self {
506
        Services {
507
            records: Default::default(),
508
            config: Arc::new(config),
509
        }
510
    }
511

            
512
    /// Connect to a hidden service
513
    // We *do* drop guard.  There is *one* await point, just after drop(guard).
514
    #[instrument(skip_all, level = "trace")]
515
40
    pub(crate) async fn get_or_launch_connection(
516
40
        connector: &HsClientConnector<impl Runtime, D>,
517
40
        netdir: &Arc<NetDir>,
518
40
        hs_id: HsId,
519
40
        isolation: Box<dyn Isolation>,
520
40
        secret_keys: HsClientSecretKeys,
521
40
    ) -> Result<Arc<D::DataTunnel>, ConnError> {
522
18
        let blank_state = || ServiceState::blank(&connector.runtime);
523

            
524
        let mut rechecks = 0..MAX_RECHECKS;
525

            
526
68
        let mut obtain = |table_index, guard| {
527
68
            obtain_circuit_or_continuation_info(
528
68
                connector,
529
68
                netdir,
530
68
                &hs_id,
531
68
                &secret_keys,
532
68
                table_index,
533
68
                &mut rechecks,
534
68
                guard,
535
            )
536
68
        };
537

            
538
        let mut got;
539
        let table_index;
540
        {
541
            let mut guard = connector.services()?;
542
            let services = &mut *guard;
543

            
544
            trace!("HS conn get_or_launch: {hs_id:?} {isolation:?} {secret_keys:?}");
545
            //trace!("HS conn services: {services:?}");
546

            
547
            table_index =
548
                services
549
                    .records
550
                    .index_or_insert_with(&hs_id, &secret_keys, isolation, blank_state);
551

            
552
            let guard = guard;
553
            got = obtain(table_index, guard);
554
        }
555
        loop {
556
            // The parts of this loop which run after a `Left` is returned
557
            // logically belong in the case in `obtain_circuit_or_continuation_info`
558
            // for `ServiceState::Working`, where that function decides we need to wait.
559
            // This code has to be out here to help the compiler's drop tracking.
560
            {
561
                // Block to scope the acquisition of `error`, a guard
562
                // for the mutex-protected error field in the state,
563
                // and, for neatness, barrier_recv.
564

            
565
                let (error, mut barrier_recv) = match got? {
566
                    Right(ret) => return Ok(ret),
567
                    Left(continuation) => continuation,
568
                };
569

            
570
                barrier_recv.recv().await;
571

            
572
                let error = error
573
                    .lock()
574
                    .map_err(|_| internal!("Working error poisoned"))?;
575
                if let Some(error) = &*error {
576
                    return Err(error.clone());
577
                }
578
            }
579

            
580
            let guard = connector.services()?;
581

            
582
            got = obtain(table_index, guard);
583
        }
584
40
    }
585

            
586
    /// Perform housekeeping - delete data we aren't interested in any more
587
20
    pub(crate) fn run_housekeeping(&mut self, now: Instant) {
588
20
        self.expire_old_data(now);
589
20
    }
590

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

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

            
644
26
        let mut maybe_wait = calculate_expiry_wait(last_used, now);
645
26
        let (mut schedule, handle) = TaskSchedule::new(connector.runtime.clone());
646
26
        let () = connector.runtime.spawn({
647
26
            let connector = connector.clone();
648
26
            async move {
649
                // This loop is slightly odd.  The wait ought naturally to be at the end,
650
                // but that would mean a useless re-lock and re-check right after creation,
651
                // or jumping into the middle of the loop.
652
                loop {
653
30
                    if let Some(yes_wait) = maybe_wait {
654
30
                        if schedule.sleep(yes_wait).await.is_err() {
655
                            // the circuit expiry task has already been canceled
656
                            break;
657
16
                        }
658
                    }
659
                    // If it's None, we can't rely on that to say we should expire it,
660
                    // since that information crossed a time when we didn't hold the lock.
661

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

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

            
713
    /// Mock state
714
    type MockGlobalState: Clone + Sync + Send + 'static;
715

            
716
    /// Connect
717
    async fn connect<R: Runtime>(
718
        connector: &HsClientConnector<R, Self>,
719
        netdir: Arc<NetDir>,
720
        config: Arc<Config>,
721
        hsid: HsId,
722
        data: &mut Self,
723
        secret_keys: HsClientSecretKeys,
724
    ) -> Result<Self::DataTunnel, ConnError>;
725

            
726
    /// Is circuit OK?  Ie, not `.is_closing()`.
727
    fn tunnel_is_ok(tunnel: &Self::DataTunnel) -> bool;
728
}
729

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

            
759
    use ConnError as E;
760

            
761
    #[derive(Debug, Default)]
762
    struct MockData {
763
        connect_called: usize,
764
    }
765

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

            
769
    #[derive(Debug, Clone)]
770
    struct MockGlobalState {
771
        // things will appear here when we have more sophisticated tests
772
        give: postage::watch::Receiver<MockGive>,
773
    }
774

            
775
    #[derive(Clone, Educe)]
776
    #[educe(Debug)]
777
    struct MockTunnel {
778
        #[educe(Debug(method = "debug_arc_mutex"))]
779
        ok: Arc<Mutex<bool>>,
780
        connect_called: usize,
781
    }
782

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

            
794
    impl PartialEq for MockTunnel {
795
        fn eq(&self, other: &MockTunnel) -> bool {
796
            Arc::ptr_eq(&self.ok, &other.ok)
797
        }
798
    }
799

            
800
    impl MockTunnel {
801
        fn new(connect_called: usize) -> Self {
802
            let ok = Arc::new(Mutex::new(true));
803
            MockTunnel { ok, connect_called }
804
        }
805
    }
806

            
807
    #[async_trait]
808
    impl MockableConnectorData for MockData {
809
        type DataTunnel = MockTunnel;
810
        type MockGlobalState = MockGlobalState;
811

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

            
837
        fn tunnel_is_ok(circuit: &Self::DataTunnel) -> bool {
838
            *circuit.ok.lock().unwrap()
839
        }
840
    }
841

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

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

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

            
899
    #[allow(clippy::unnecessary_wraps)]
900
    fn mk_isol(s: &str) -> Option<NarrowableIsolation> {
901
        Some(NarrowableIsolation(s.into()))
902
    }
903

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

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

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

            
944
    #[test]
945
    #[traced_test]
946
    fn simple() {
947
        test_with_one_runtime!(|runtime| async {
948
            let (hsconn, keys, _give_send) = mk_hsconn(runtime);
949

            
950
            let circuit = launch_one(&hsconn, 0, &keys, None).await.unwrap();
951
            eprintln!("{:?}", circuit);
952
        });
953
    }
954

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

            
965
            let (hsconn, keys, _give_send) = mk_hsconn(runtime.clone());
966

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

            
981
            // make circuit1
982
            let circuit1 = launch_one(&hsconn, 0, &keys, None).await.unwrap();
983

            
984
            // expire it
985
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE + TIMEOUT_SLOP).await;
986

            
987
            // make circuit2 (a)
988
            let circuit2a = launch_one(&hsconn, 0, &keys, None).await.unwrap();
989
            assert_ne!(circuit1, circuit2a);
990

            
991
            // nearly expire it, then reuse it
992
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE - TIMEOUT_SLOP).await;
993
            let circuit2b = launch_one(&hsconn, 0, &keys, None).await.unwrap();
994
            assert_eq!(circuit2a, circuit2b);
995

            
996
            // nearly expire it again, then reuse it
997
            advance(RETAIN_CIRCUIT_AFTER_LAST_USE - TIMEOUT_SLOP).await;
998
            let circuit2c = launch_one(&hsconn, 0, &keys, None).await.unwrap();
999
            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);
        });
    }
}