1
#![cfg_attr(docsrs, feature(doc_cfg))]
2
#![doc = include_str!("../README.md")]
3
// @@ begin lint list maintained by maint/add_warning @@
4
#![allow(renamed_and_removed_lints)] // @@REMOVE_WHEN(ci_arti_stable)
5
#![allow(unknown_lints)] // @@REMOVE_WHEN(ci_arti_nightly)
6
#![warn(missing_docs)]
7
#![warn(noop_method_call)]
8
#![warn(unreachable_pub)]
9
#![warn(clippy::all)]
10
#![deny(clippy::await_holding_lock)]
11
#![deny(clippy::cargo_common_metadata)]
12
#![deny(clippy::cast_lossless)]
13
#![deny(clippy::checked_conversions)]
14
#![allow(clippy::cognitive_complexity)] // See arti#2556
15
#![deny(clippy::debug_assert_with_mut_call)]
16
#![deny(clippy::exhaustive_enums)]
17
#![deny(clippy::exhaustive_structs)]
18
#![deny(clippy::expl_impl_clone_on_copy)]
19
#![deny(clippy::fallible_impl_from)]
20
#![deny(clippy::implicit_clone)]
21
#![deny(clippy::large_stack_arrays)]
22
#![warn(clippy::manual_ok_or)]
23
#![deny(clippy::missing_docs_in_private_items)]
24
#![warn(clippy::needless_borrow)]
25
#![warn(clippy::needless_pass_by_value)]
26
#![warn(clippy::option_option)]
27
#![warn(clippy::or_fun_call)]
28
#![deny(clippy::print_stderr)]
29
#![deny(clippy::print_stdout)]
30
#![warn(clippy::rc_buffer)]
31
#![deny(clippy::ref_option_ref)]
32
#![warn(clippy::semicolon_if_nothing_returned)]
33
#![warn(clippy::trait_duplication_in_bounds)]
34
#![deny(clippy::unchecked_time_subtraction)]
35
#![deny(clippy::unnecessary_wraps)]
36
#![warn(clippy::unseparated_literal_suffix)]
37
#![deny(clippy::unwrap_used)]
38
#![deny(clippy::mod_module_files)]
39
#![allow(clippy::let_unit_value)] // This can reasonably be done for explicitness
40
#![allow(clippy::uninlined_format_args)]
41
#![allow(clippy::significant_drop_in_scrutinee)] // arti/-/merge_requests/588/#note_2812945
42
#![allow(clippy::result_large_err)] // temporary workaround for arti#587
43
#![allow(clippy::needless_raw_string_hashes)] // complained-about code is fine, often best
44
#![allow(clippy::needless_lifetimes)] // See arti#1765
45
#![allow(mismatched_lifetime_syntaxes)] // temporary workaround for arti#2060
46
#![allow(clippy::collapsible_if)] // See arti#2342
47
#![deny(clippy::unused_async)]
48
#![deny(clippy::string_slice)] // See arti#2571
49
#![allow(recursion_depth_exceeding_limit)] // arti#2715, rust/issues/159228
50
//! <!-- @@ end lint list maintained by maint/add_warning @@ -->
51

            
52
// This clippy lint produces a false positive on `use strum`, below.
53
// Attempting to apply the lint to just the use statement fails to suppress
54
// this lint and instead produces another lint about a useless clippy attribute.
55
#![allow(clippy::single_component_path_imports)]
56

            
57
mod bootstrap;
58
pub mod config;
59
mod docid;
60
mod docmeta;
61
mod err;
62
mod event;
63
mod shared_ref;
64
mod state;
65
mod storage;
66

            
67
#[cfg(feature = "dir-plugin")]
68
mod as_plugin;
69
#[cfg(feature = "bridge-client")]
70
pub mod bridgedesc;
71
#[cfg(feature = "dirfilter")]
72
pub mod filter;
73

            
74
use crate::docid::{CacheUsage, ClientRequest, DocQuery};
75
use crate::err::BootstrapAction;
76
#[cfg(not(feature = "experimental-api"))]
77
use crate::shared_ref::SharedMutArc;
78
#[cfg(feature = "experimental-api")]
79
pub use crate::shared_ref::SharedMutArc;
80
use crate::storage::{DynStore, Store};
81
use bootstrap::AttemptId;
82
use event::DirProgress;
83
use postage::watch;
84
use scopeguard::ScopeGuard;
85
use tor_circmgr::CircMgr;
86
use tor_dirclient::SourceInfo;
87
use tor_dircommon::config::DirTolerance;
88
use tor_error::{info_report, into_internal, warn_report};
89
use tor_netdir::params::NetParameters;
90
use tor_netdir::{DirEvent, MdReceiver, NetDir, NetDirProvider};
91

            
92
use async_trait::async_trait;
93
use futures::stream::BoxStream;
94
use oneshot_fused_workaround as oneshot;
95
use tor_netdoc::doc::netstatus::ProtoStatuses;
96
use tor_rtcompat::scheduler::{TaskHandle, TaskSchedule};
97
use tor_rtcompat::{Runtime, SpawnExt};
98
use tracing::{debug, info, instrument, trace, warn};
99
use web_time_compat::SystemTimeExt;
100

            
101
use std::marker::PhantomData;
102
use std::sync::atomic::{AtomicBool, Ordering};
103
use std::sync::{Arc, Mutex};
104
use std::time::Duration;
105
use std::{collections::HashMap, sync::Weak};
106
use std::{fmt::Debug, time::SystemTime};
107

            
108
use crate::state::{DirState, NetDirChange};
109
pub use config::DirMgrConfig;
110
pub use docid::DocId;
111
pub use err::Error;
112
pub use event::{DirBlockage, DirBootstrapEvents, DirBootstrapStatus};
113
pub use storage::DocumentText;
114
pub use tor_dircommon::fallback::{FallbackDir, FallbackDirBuilder};
115
pub use tor_netdir::Timeliness;
116

            
117
#[cfg(feature = "dir-plugin")]
118
pub use as_plugin::DirPlugin;
119

            
120
/// Re-export of `strum` crate for use by an internal macro
121
use strum;
122

            
123
/// A Result as returned by this crate.
124
pub type Result<T> = std::result::Result<T, Error>;
125

            
126
/// Storage manager used by [`DirMgr`] and
127
/// [`BridgeDescMgr`](bridgedesc::BridgeDescMgr)
128
///
129
/// Internally, this wraps up a sqlite database.
130
///
131
/// This is a handle, which is cheap to clone; clones share state.
132
#[derive(Clone)]
133
pub struct DirMgrStore<R: Runtime> {
134
    /// The actual store
135
    pub(crate) store: Arc<Mutex<crate::DynStore>>,
136

            
137
    /// Be parameterized by Runtime even though we don't use it right now
138
    pub(crate) runtime: PhantomData<R>,
139
}
140

            
141
impl<R: Runtime> DirMgrStore<R> {
142
    /// Open the storage, according to the specified configuration
143
18
    pub fn new(config: &DirMgrConfig, runtime: R, offline: bool) -> Result<Self> {
144
18
        let store = Arc::new(Mutex::new(config.open_store(offline)?));
145
18
        drop(runtime);
146
18
        let runtime = PhantomData;
147
18
        Ok(DirMgrStore { store, runtime })
148
18
    }
149
}
150

            
151
/// Trait for DirMgr implementations
152
#[async_trait]
153
pub trait DirProvider: NetDirProvider {
154
    /// Try to change our configuration to `new_config`.
155
    ///
156
    /// Actual behavior will depend on the value of `how`.
157
    fn reconfigure(
158
        &self,
159
        new_config: &DirMgrConfig,
160
        how: tor_config::Reconfigure,
161
    ) -> std::result::Result<(), tor_config::ReconfigureError>;
162

            
163
    /// Bootstrap a `DirProvider` that hasn't been bootstrapped yet.
164
    async fn bootstrap(&self) -> Result<()>;
165

            
166
    /// Return a stream of [`DirBootstrapStatus`] events to tell us about changes
167
    /// in the latest directory's bootstrap status.
168
    ///
169
    /// Note that this stream can be lossy: the caller will not necessarily
170
    /// observe every event on the stream
171
    fn bootstrap_events(&self) -> BoxStream<'static, DirBootstrapStatus>;
172

            
173
    /// Return a [`TaskHandle`] that can be used to manage the download process.
174
    fn download_task_handle(&self) -> Option<TaskHandle> {
175
        None
176
    }
177
}
178

            
179
// NOTE(eta): We can't implement this for Arc<DirMgr<R>> due to trait coherence rules, so instead
180
//            there's a blanket impl for Arc<T> in tor-netdir.
181
impl<R: Runtime> NetDirProvider for DirMgr<R> {
182
1
    fn netdir(&self, timeliness: Timeliness) -> tor_netdir::Result<Arc<NetDir>> {
183
        use tor_netdir::Error as NetDirError;
184
1
        let netdir = self.netdir.get().ok_or(NetDirError::NoInfo)?;
185
        let lifetime = match timeliness {
186
            Timeliness::Strict => netdir.lifetime().clone(),
187
            Timeliness::Timely => self
188
                .config
189
                .get()
190
                .tolerance
191
                .extend_lifetime(netdir.lifetime()),
192
            Timeliness::Unchecked => return Ok(netdir),
193
        };
194
        // TODO #2384 -- we have a runtime here; we should use it.
195
        let now = SystemTime::get();
196
        if lifetime.valid_after() > now {
197
            Err(NetDirError::DirNotYetValid)
198
        } else if lifetime.valid_until() < now {
199
            Err(NetDirError::DirExpired)
200
        } else {
201
            Ok(netdir)
202
        }
203
1
    }
204

            
205
    fn events(&self) -> BoxStream<'static, DirEvent> {
206
        Box::pin(self.events.subscribe())
207
    }
208

            
209
    fn params(&self) -> Arc<dyn AsRef<tor_netdir::params::NetParameters>> {
210
        if let Some(netdir) = self.netdir.get() {
211
            // We have a directory, so we'd like to give it out for its
212
            // parameters.
213
            //
214
            // We do this even if the directory is expired, since parameters
215
            // don't really expire on any plausible timescale.
216
            netdir
217
        } else {
218
            // We have no directory, so we'll give out the default parameters as
219
            // modified by the provided override_net_params configuration.
220
            //
221
            self.default_parameters
222
                .lock()
223
                .expect("Poisoned lock")
224
                .clone()
225
        }
226
        // TODO(nickm): If we felt extremely clever, we could add a third case
227
        // where, if we have a pending directory with a validated consensus, we
228
        // give out that consensus's network parameters even if we _don't_ yet
229
        // have a full directory.  That's significant refactoring, though, for
230
        // an unclear amount of benefit.
231
    }
232

            
233
    fn protocol_statuses(&self) -> Option<(SystemTime, Arc<ProtoStatuses>)> {
234
        self.protocols.lock().expect("Poisoned lock").clone()
235
    }
236
}
237

            
238
#[async_trait]
239
impl<R: Runtime> DirProvider for Arc<DirMgr<R>> {
240
    fn reconfigure(
241
        &self,
242
        new_config: &DirMgrConfig,
243
        how: tor_config::Reconfigure,
244
    ) -> std::result::Result<(), tor_config::ReconfigureError> {
245
        DirMgr::reconfigure(self, new_config, how)
246
    }
247

            
248
    #[instrument(level = "trace", skip_all)]
249
    async fn bootstrap(&self) -> Result<()> {
250
        DirMgr::bootstrap(self).await
251
    }
252

            
253
    fn bootstrap_events(&self) -> BoxStream<'static, DirBootstrapStatus> {
254
        Box::pin(DirMgr::bootstrap_events(self))
255
    }
256

            
257
    fn download_task_handle(&self) -> Option<TaskHandle> {
258
        Some(self.task_handle.clone())
259
    }
260
}
261

            
262
/// A directory manager to download, fetch, and cache a Tor directory.
263
///
264
/// A DirMgr can operate in three modes:
265
///   * In **offline** mode, it only reads from the cache, and can
266
///     only read once.
267
///   * In **read-only** mode, it reads from the cache, but checks
268
///     whether it can acquire an associated lock file.  If it can, then
269
///     it enters read-write mode.  If not, it checks the cache
270
///     periodically for new information.
271
///   * In **read-write** mode, it knows that no other process will be
272
///     writing to the cache, and it takes responsibility for fetching
273
///     data from the network and updating the directory with new
274
///     directory information.
275
pub struct DirMgr<R: Runtime> {
276
    /// Configuration information: where to find directories, how to
277
    /// validate them, and so on.
278
    config: tor_config::MutCfg<DirMgrConfig>,
279
    /// Handle to our sqlite cache.
280
    // TODO(nickm): I'd like to use an rwlock, but that's not feasible, since
281
    // rusqlite::Connection isn't Sync.
282
    // TODO is needed?
283
    store: Arc<Mutex<DynStore>>,
284
    /// Our latest sufficiently bootstrapped directory, if we have one.
285
    ///
286
    /// We use the RwLock so that we can give this out to a bunch of other
287
    /// users, and replace it once a new directory is bootstrapped.
288
    // TODO(eta): Eurgh! This is so many Arcs! (especially considering this
289
    //            gets wrapped in an Arc)
290
    netdir: Arc<SharedMutArc<NetDir>>,
291

            
292
    /// Our latest set of recommended protocols.
293
    protocols: Mutex<Option<(SystemTime, Arc<ProtoStatuses>)>>,
294

            
295
    /// A set of network parameters to hand out when we have no directory.
296
    default_parameters: Mutex<Arc<NetParameters>>,
297

            
298
    /// A publisher handle that we notify whenever the consensus changes.
299
    events: event::FlagPublisher<DirEvent>,
300

            
301
    /// A publisher handle that we notify whenever our bootstrapping status
302
    /// changes.
303
    send_status: Mutex<watch::Sender<event::DirBootstrapStatus>>,
304

            
305
    /// A receiver handle that gets notified whenever our bootstrapping status
306
    /// changes.
307
    ///
308
    /// We don't need to keep this drained, since `postage::watch` already knows
309
    /// to discard unread events.
310
    receive_status: DirBootstrapEvents,
311

            
312
    /// A circuit manager, if this DirMgr supports downloading.
313
    circmgr: Option<Arc<CircMgr<R>>>,
314

            
315
    /// Our asynchronous runtime.
316
    runtime: R,
317

            
318
    /// Whether or not we're operating in offline mode.
319
    offline: bool,
320

            
321
    /// If we're not in offline mode, stores whether or not the `DirMgr` has attempted
322
    /// to bootstrap yet or not.
323
    ///
324
    /// This exists in order to prevent starting two concurrent bootstrap tasks.
325
    ///
326
    /// (In offline mode, this does nothing.)
327
    bootstrap_started: AtomicBool,
328

            
329
    /// A filter that gets applied to directory objects before we use them.
330
    #[cfg(feature = "dirfilter")]
331
    filter: crate::filter::FilterConfig,
332

            
333
    /// A task schedule that can be used if we're bootstrapping.  If this is
334
    /// None, then there's currently a scheduled task in progress.
335
    task_schedule: Mutex<Option<TaskSchedule<R>>>,
336

            
337
    /// A task handle that we return to anybody who needs to manage our download process.
338
    task_handle: TaskHandle,
339
}
340

            
341
/// The possible origins of a document.
342
///
343
/// Used (for example) to report where we got a document from if it fails to
344
/// parse.
345
#[derive(Debug, Clone)]
346
#[non_exhaustive]
347
pub enum DocSource {
348
    /// We loaded the document from our cache.
349
    LocalCache,
350
    /// We fetched the document from a server.
351
    DirServer {
352
        /// Information about the server we fetched the document from.
353
        source: Option<SourceInfo>,
354
    },
355
}
356

            
357
impl std::fmt::Display for DocSource {
358
1
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359
1
        match self {
360
            DocSource::LocalCache => write!(f, "local cache"),
361
1
            DocSource::DirServer { source: None } => write!(f, "directory server"),
362
            DocSource::DirServer { source: Some(info) } => write!(f, "directory server {}", info),
363
        }
364
1
    }
365
}
366

            
367
impl<R: Runtime> DirMgr<R> {
368
    /// Try to load the directory from disk, without launching any
369
    /// kind of update process.
370
    ///
371
    /// This function runs in **offline** mode: it will give an error
372
    /// if the result is not up-to-date, or not fully downloaded.
373
    ///
374
    /// In general, you shouldn't use this function in a long-running
375
    /// program; it's only suitable for command-line or batch tools.
376
    // TODO: I wish this function didn't have to be async or take a runtime.
377
    pub fn load_once(runtime: R, config: DirMgrConfig) -> Result<Arc<NetDir>> {
378
        let store = DirMgrStore::new(&config, runtime.clone(), true)?;
379
        let dirmgr = Arc::new(Self::from_config(config, runtime, store, None, true)?);
380

            
381
        // TODO: add some way to return a directory that isn't up-to-date
382
        let attempt = AttemptId::next();
383
        trace!(%attempt, "Trying to load a full directory from cache");
384
        let outcome = dirmgr.load_directory(attempt);
385
        trace!(%attempt, "Load result: {outcome:?}");
386
        let _success = outcome?;
387

            
388
        dirmgr
389
            .netdir(Timeliness::Timely)
390
            .map_err(|_| Error::DirectoryNotPresent)
391
    }
392

            
393
    /// Return a current netdir, either loading it or bootstrapping it
394
    /// as needed.
395
    ///
396
    /// Like load_once, but will try to bootstrap (or wait for another
397
    /// process to bootstrap) if we don't have an up-to-date
398
    /// bootstrapped directory.
399
    ///
400
    /// In general, you shouldn't use this function in a long-running
401
    /// program; it's only suitable for command-line or batch tools.
402
    pub async fn load_or_bootstrap_once(
403
        config: DirMgrConfig,
404
        runtime: R,
405
        store: DirMgrStore<R>,
406
        circmgr: Arc<CircMgr<R>>,
407
    ) -> Result<Arc<NetDir>> {
408
        let dirmgr = DirMgr::bootstrap_from_config(config, runtime, store, circmgr).await?;
409
        dirmgr
410
            .timely_netdir()
411
            .map_err(|_| Error::DirectoryNotPresent)
412
    }
413

            
414
    /// Create a new `DirMgr` in online mode, but don't bootstrap it yet.
415
    ///
416
    /// The `DirMgr` can be bootstrapped later with `bootstrap`.
417
    pub fn create_unbootstrapped(
418
        config: DirMgrConfig,
419
        runtime: R,
420
        store: DirMgrStore<R>,
421
        circmgr: Arc<CircMgr<R>>,
422
    ) -> Result<Arc<Self>> {
423
        Ok(Arc::new(DirMgr::from_config(
424
            config,
425
            runtime,
426
            store,
427
            Some(circmgr),
428
            false,
429
        )?))
430
    }
431

            
432
    /// Bootstrap a `DirMgr` created in online mode that hasn't been bootstrapped yet.
433
    ///
434
    /// This function will not return until the directory is bootstrapped enough to build circuits.
435
    /// It will also launch a background task that fetches any missing information, and that
436
    /// replaces the directory when a new one is available.
437
    ///
438
    /// This function is intended to be used together with `create_unbootstrapped`. There is no
439
    /// need to call this function otherwise.
440
    ///
441
    /// If bootstrapping has already successfully taken place, returns early with success.
442
    ///
443
    /// # Errors
444
    ///
445
    /// Returns an error if bootstrapping fails. If the error is [`Error::CantAdvanceState`],
446
    /// it may be possible to successfully bootstrap later on by calling this function again.
447
    ///
448
    /// # Panics
449
    ///
450
    /// Panics if the `DirMgr` passed to this function was not created in online mode, such as
451
    /// via `load_once`.
452
    #[instrument(level = "trace", skip_all)]
453
    pub async fn bootstrap(self: &Arc<Self>) -> Result<()> {
454
        if self.offline {
455
            return Err(Error::OfflineMode);
456
        }
457

            
458
        // The semantics of this are "attempt to replace a 'false' value with 'true'.
459
        // If the value in bootstrap_started was not 'false' when the attempt was made, returns
460
        // `Err`; this means another bootstrap attempt is in progress or has completed, so we
461
        // return early.
462

            
463
        // NOTE(eta): could potentially weaken the `Ordering` here in future
464
        if self
465
            .bootstrap_started
466
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
467
            .is_err()
468
        {
469
            debug!("Attempted to bootstrap twice; ignoring.");
470
            return Ok(());
471
        }
472

            
473
        // Use a RAII guard to reset `bootstrap_started` to `false` if we return early without
474
        // completing bootstrap.
475
        let reset_bootstrap_started = scopeguard::guard(&self.bootstrap_started, |v| {
476
            v.store(false, Ordering::SeqCst);
477
        });
478

            
479
        let schedule = {
480
            let sched = self.task_schedule.lock().expect("poisoned lock").take();
481
            match sched {
482
                Some(sched) => sched,
483
                None => {
484
                    debug!("Attempted to bootstrap twice; ignoring.");
485
                    return Ok(());
486
                }
487
            }
488
        };
489

            
490
        // Try to load from the cache.
491
        let attempt_id = AttemptId::next();
492
        trace!(attempt=%attempt_id, "Starting to bootstrap directory");
493
        let have_directory = self.load_directory(attempt_id)?;
494

            
495
        let (mut sender, receiver) = if have_directory {
496
            info!("Loaded a good directory from cache.");
497
            (None, None)
498
        } else {
499
            info!("Didn't get usable directory from cache.");
500
            let (sender, receiver) = oneshot::channel();
501
            (Some(sender), Some(receiver))
502
        };
503

            
504
        // Whether we loaded or not, we now start downloading.
505
        let dirmgr_weak = Arc::downgrade(self);
506
        self.runtime
507
            .spawn(async move {
508
                // Use an RAII guard to make sure that when this task exits, the
509
                // TaskSchedule object is put back.
510
                //
511
                // TODO(nick): Putting the schedule back isn't actually useful
512
                // if the task exits _after_ we've bootstrapped for the first
513
                // time, because of how bootstrap_started works.
514
                let mut schedule = scopeguard::guard(schedule, |schedule| {
515
                    if let Some(dm) = Weak::upgrade(&dirmgr_weak) {
516
                        *dm.task_schedule.lock().expect("poisoned lock") = Some(schedule);
517
                    }
518
                });
519

            
520
                // Don't warn when these are Error::ManagerDropped: that
521
                // means that the DirMgr has been shut down.
522
                if let Err(e) =
523
                    Self::reload_until_owner(&dirmgr_weak, &mut schedule, attempt_id, &mut sender)
524
                        .await
525
                {
526
                    match e {
527
                        Error::ManagerDropped => {}
528
                        _ => warn_report!(e, "Unrecovered error while waiting for bootstrap",),
529
                    }
530
                } else if let Err(e) =
531
                    Self::download_forever(dirmgr_weak.clone(), &mut schedule, attempt_id, sender)
532
                        .await
533
                {
534
                    match e {
535
                        Error::ManagerDropped => {}
536
                        _ => warn_report!(e, "Unrecovered error while downloading"),
537
                    }
538
                }
539
            })
540
            .map_err(|e| Error::from_spawn("directory updater task", e))?;
541

            
542
        if let Some(receiver) = receiver {
543
            match receiver.await {
544
                Ok(()) => {
545
                    info!("We have enough information to build circuits.");
546
                    // Disarm the RAII guard, since we succeeded.  Now bootstrap_started will remain true.
547
                    let _ = ScopeGuard::into_inner(reset_bootstrap_started);
548
                }
549
                Err(_) => {
550
                    warn!("Bootstrapping task exited before finishing.");
551
                    return Err(Error::CantAdvanceState);
552
                }
553
            }
554
        }
555
        Ok(())
556
    }
557

            
558
    /// Returns `true` if a bootstrap attempt is in progress, or successfully completed.
559
    pub fn bootstrap_started(&self) -> bool {
560
        self.bootstrap_started.load(Ordering::SeqCst)
561
    }
562

            
563
    /// Return a new directory manager from a given configuration,
564
    /// bootstrapping from the network as necessary.
565
    #[instrument(level = "trace", skip_all)]
566
    pub async fn bootstrap_from_config(
567
        config: DirMgrConfig,
568
        runtime: R,
569
        store: DirMgrStore<R>,
570
        circmgr: Arc<CircMgr<R>>,
571
    ) -> Result<Arc<Self>> {
572
        let dirmgr = Self::create_unbootstrapped(config, runtime, store, circmgr)?;
573

            
574
        dirmgr.bootstrap().await?;
575

            
576
        Ok(dirmgr)
577
    }
578

            
579
    /// Try forever to either lock the storage (and thereby become the
580
    /// owner), or to reload the database.
581
    ///
582
    /// If we have begin to have a bootstrapped directory, send a
583
    /// message using `on_complete`.
584
    ///
585
    /// If we eventually become the owner, return Ok().
586
    async fn reload_until_owner(
587
        weak: &Weak<Self>,
588
        schedule: &mut TaskSchedule<R>,
589
        attempt_id: AttemptId,
590
        on_complete: &mut Option<oneshot::Sender<()>>,
591
    ) -> Result<()> {
592
        let mut logged = false;
593
        let mut bootstrapped;
594
        {
595
            let dirmgr = upgrade_weak_ref(weak)?;
596
            bootstrapped = dirmgr.netdir.get().is_some();
597
        }
598

            
599
        loop {
600
            {
601
                let dirmgr = upgrade_weak_ref(weak)?;
602
                trace!("Trying to take ownership of the directory cache lock");
603
                if dirmgr.try_upgrade_to_readwrite()? {
604
                    // We now own the lock!  (Maybe we owned it before; the
605
                    // upgrade_to_readwrite() function is idempotent.)  We can
606
                    // do our own bootstrapping.
607
                    if logged {
608
                        info!(
609
                            "The previous owning process has given up the lock. We are now in charge of managing the directory."
610
                        );
611
                    }
612
                    return Ok(());
613
                }
614
            }
615

            
616
            if !logged {
617
                logged = true;
618
                if bootstrapped {
619
                    info!("Another process is managing the directory. We'll use its cache.");
620
                } else {
621
                    info!(
622
                        "Another process is bootstrapping the directory. Waiting till it finishes or exits."
623
                    );
624
                }
625
            }
626

            
627
            // We don't own the lock.  Somebody else owns the cache.  They
628
            // should be updating it.  Wait a bit, then try again.
629
            let pause = if bootstrapped {
630
                std::time::Duration::new(120, 0)
631
            } else {
632
                std::time::Duration::new(5, 0)
633
            };
634
            schedule.sleep(pause).await?;
635
            // TODO: instead of loading the whole thing we should have a
636
            // database entry that says when the last update was, or use
637
            // our state functions.
638
            {
639
                let dirmgr = upgrade_weak_ref(weak)?;
640
                trace!("Trying to load from the directory cache");
641
                if dirmgr.load_directory(attempt_id)? {
642
                    // Successfully loaded a bootstrapped directory.
643
                    if let Some(send_done) = on_complete.take() {
644
                        let _ = send_done.send(());
645
                    }
646
                    if !bootstrapped {
647
                        info!("The directory is now bootstrapped.");
648
                    }
649
                    bootstrapped = true;
650
                }
651
            }
652
        }
653
    }
654

            
655
    /// Try to fetch our directory info and keep it updated, indefinitely.
656
    ///
657
    /// If we have begin to have a bootstrapped directory, send a
658
    /// message using `on_complete`.
659
    #[instrument(level = "trace", skip_all)]
660
    async fn download_forever(
661
        weak: Weak<Self>,
662
        schedule: &mut TaskSchedule<R>,
663
        mut attempt_id: AttemptId,
664
        mut on_complete: Option<oneshot::Sender<()>>,
665
    ) -> Result<()> {
666
        let mut state: Box<dyn DirState> = {
667
            let dirmgr = upgrade_weak_ref(&weak)?;
668
            Box::new(state::GetConsensusState::new(
669
                dirmgr.runtime.clone(),
670
                dirmgr.config.get(),
671
                CacheUsage::CacheOkay,
672
                Some(dirmgr.netdir.clone()),
673
                #[cfg(feature = "dirfilter")]
674
                dirmgr
675
                    .filter
676
                    .clone()
677
                    .unwrap_or_else(|| Arc::new(crate::filter::NilFilter)),
678
            ))
679
        };
680

            
681
        trace!("Entering download loop.");
682

            
683
        loop {
684
            let mut usable = false;
685

            
686
            let retry_config = {
687
                let dirmgr = upgrade_weak_ref(&weak)?;
688
                // TODO(nickm): instead of getting this every time we loop, it
689
                // might be a good idea to refresh it with each attempt, at
690
                // least at the point of checking the number of attempts.
691
                dirmgr.config.get().schedule.retry_bootstrap()
692
            };
693
            let mut retry_delay = retry_config.schedule();
694

            
695
            'retry_attempt: for try_num in retry_config.attempts() {
696
                trace!(attempt=%attempt_id, ?try_num, "Trying to download a directory.");
697
                let outcome = bootstrap::download(
698
                    Weak::clone(&weak),
699
                    &mut state,
700
                    schedule,
701
                    attempt_id,
702
                    &mut on_complete,
703
                )
704
                .await;
705
                trace!(attempt=%attempt_id, ?try_num, ?outcome, "Download is over.");
706

            
707
                if let Err(err) = outcome {
708
                    if state.is_ready(Readiness::Usable) {
709
                        usable = true;
710
                        info_report!(
711
                            err,
712
                            "Unable to completely download a directory. (Nevertheless, the directory is usable, so we'll pause for now)"
713
                        );
714
                        break 'retry_attempt;
715
                    }
716

            
717
                    match err.bootstrap_action() {
718
                        BootstrapAction::Nonfatal => {
719
                            return Err(into_internal!(
720
                                "Nonfatal error should not have propagated here"
721
                            )(err)
722
                            .into());
723
                        }
724
                        BootstrapAction::Reset => {}
725
                        BootstrapAction::Fatal => return Err(err),
726
                    }
727

            
728
                    let delay = retry_delay.next_delay(&mut rand::rng());
729
                    warn_report!(
730
                        err,
731
                        "Unable to download a usable directory. (We will restart in {})",
732
                        humantime::format_duration(delay),
733
                    );
734
                    {
735
                        let dirmgr = upgrade_weak_ref(&weak)?;
736
                        dirmgr.note_reset(attempt_id);
737
                    }
738
                    schedule.sleep(delay).await?;
739
                    state = state.reset();
740
                } else {
741
                    info!(attempt=%attempt_id, "Directory is complete.");
742
                    usable = true;
743
                    break 'retry_attempt;
744
                }
745
            }
746

            
747
            if !usable {
748
                // we ran out of attempts.
749
                warn!(
750
                    "We failed {} times to bootstrap a directory. We're going to give up.",
751
                    retry_config.n_attempts()
752
                );
753
                return Err(Error::CantAdvanceState);
754
            } else {
755
                // Report success, if appropriate.
756
                if let Some(send_done) = on_complete.take() {
757
                    let _ = send_done.send(());
758
                }
759
            }
760

            
761
            let reset_at = state.reset_time();
762
            match reset_at {
763
                Some(t) => {
764
                    trace!("Sleeping until {}", time::OffsetDateTime::from(t));
765
                    schedule.sleep_until_wallclock(t).await?;
766
                }
767
                None => return Ok(()),
768
            }
769
            attempt_id = bootstrap::AttemptId::next();
770
            trace!(attempt=%attempt_id, "Beginning new attempt to bootstrap directory");
771
            state = state.reset();
772
        }
773
    }
774

            
775
    /// Get a reference to the circuit manager, if we have one.
776
1
    fn circmgr(&self) -> Result<Arc<CircMgr<R>>> {
777
1
        self.circmgr.clone().ok_or(Error::NoDownloadSupport)
778
1
    }
779

            
780
    /// Try to change our configuration to `new_config`.
781
    ///
782
    /// Actual behavior will depend on the value of `how`.
783
    pub fn reconfigure(
784
        &self,
785
        new_config: &DirMgrConfig,
786
        how: tor_config::Reconfigure,
787
    ) -> std::result::Result<(), tor_config::ReconfigureError> {
788
        let config = self.config.get();
789
        // We don't support changing these: doing so basically would require us
790
        // to abort all our in-progress downloads, since they might be based on
791
        // no-longer-viable information.
792
        // NOTE: keep this in sync with the behaviour of `DirMgrConfig::update_from_config`
793
        if new_config.cache_dir != config.cache_dir {
794
            how.cannot_change("storage.cache_dir")?;
795
        }
796
        if new_config.cache_trust != config.cache_trust {
797
            how.cannot_change("storage.permissions")?;
798
        }
799
        if new_config.authorities() != config.authorities() {
800
            how.cannot_change("network.authorities")?;
801
        }
802

            
803
        if how == tor_config::Reconfigure::CheckAllOrNothing {
804
            return Ok(());
805
        }
806

            
807
        let params_changed = new_config.override_net_params != config.override_net_params;
808

            
809
        self.config
810
            .map_and_replace(|cfg| cfg.update_from_config(new_config));
811

            
812
        if params_changed {
813
            let _ignore_err = self.netdir.mutate(|netdir| {
814
                netdir.replace_overridden_parameters(&new_config.override_net_params);
815
                Ok(())
816
            });
817
            {
818
                let mut params = self.default_parameters.lock().expect("lock failed");
819
                *params = Arc::new(NetParameters::from_map(&new_config.override_net_params));
820
            }
821

            
822
            // (It's okay to ignore the error, since it just means that there
823
            // was no current netdir.)
824
            self.events.publish(DirEvent::NewConsensus);
825
        }
826

            
827
        Ok(())
828
    }
829

            
830
    /// Return a stream of [`DirBootstrapStatus`] events to tell us about changes
831
    /// in the latest directory's bootstrap status.
832
    ///
833
    /// Note that this stream can be lossy: the caller will not necessarily
834
    /// observe every event on the stream
835
    pub fn bootstrap_events(&self) -> event::DirBootstrapEvents {
836
        self.receive_status.clone()
837
    }
838

            
839
    /// Replace the latest status with `progress` and broadcast to anybody
840
    /// watching via a [`DirBootstrapEvents`] stream.
841
15
    fn update_progress(&self, attempt_id: AttemptId, progress: DirProgress) {
842
        // TODO(nickm): can I kill off this lock by having something else own the sender?
843
15
        let mut sender = self.send_status.lock().expect("poisoned lock");
844
15
        let mut status = sender.borrow_mut();
845

            
846
15
        status.update_progress(attempt_id, progress);
847
15
    }
848

            
849
    /// Update our status tracker to note that some number of errors has
850
    /// occurred.
851
    fn note_errors(&self, attempt_id: AttemptId, n_errors: usize) {
852
        if n_errors == 0 {
853
            return;
854
        }
855
        let mut sender = self.send_status.lock().expect("poisoned lock");
856
        let mut status = sender.borrow_mut();
857

            
858
        status.note_errors(attempt_id, n_errors);
859
    }
860

            
861
    /// Update our status tracker to note that we've needed to reset our download attempt.
862
    fn note_reset(&self, attempt_id: AttemptId) {
863
        let mut sender = self.send_status.lock().expect("poisoned lock");
864
        let mut status = sender.borrow_mut();
865

            
866
        status.note_reset(attempt_id);
867
    }
868

            
869
    /// Try to make this a directory manager with read-write access to its
870
    /// storage.
871
    ///
872
    /// Return true if we got the lock, or if we already had it.
873
    ///
874
    /// Return false if another process has the lock
875
    fn try_upgrade_to_readwrite(&self) -> Result<bool> {
876
        self.store
877
            .lock()
878
            .expect("Directory storage lock poisoned")
879
            .upgrade_to_readwrite()
880
    }
881

            
882
    /// Return a reference to the store, if it is currently read-write.
883
    #[cfg(test)]
884
2
    fn store_if_rw(&self) -> Option<&Mutex<DynStore>> {
885
2
        let rw = !self
886
2
            .store
887
2
            .lock()
888
2
            .expect("Directory storage lock poisoned")
889
2
            .is_readonly();
890
        // A race-condition is possible here, but I believe it's harmless.
891
2
        if rw { Some(&self.store) } else { None }
892
2
    }
893

            
894
    /// Construct a DirMgr from a DirMgrConfig.
895
    ///
896
    /// If `offline` is set, opens the SQLite store read-only and sets the offline flag in the
897
    /// returned manager.
898
    #[allow(clippy::unnecessary_wraps)] // API compat and future-proofing
899
7
    fn from_config(
900
7
        config: DirMgrConfig,
901
7
        runtime: R,
902
7
        store: DirMgrStore<R>,
903
7
        circmgr: Option<Arc<CircMgr<R>>>,
904
7
        offline: bool,
905
7
    ) -> Result<Self> {
906
7
        let netdir = Arc::new(SharedMutArc::new());
907
7
        let events = event::FlagPublisher::new();
908
7
        let default_parameters = NetParameters::from_map(&config.override_net_params);
909
7
        let default_parameters = Mutex::new(Arc::new(default_parameters));
910

            
911
7
        let (send_status, receive_status) = postage::watch::channel();
912
7
        let send_status = Mutex::new(send_status);
913
7
        let receive_status = DirBootstrapEvents {
914
7
            inner: receive_status,
915
7
        };
916
        #[cfg(feature = "dirfilter")]
917
7
        let filter = config.extensions.filter.clone();
918

            
919
        // We create these early so the client code can access task_handle before bootstrap() returns.
920
7
        let (task_schedule, task_handle) = TaskSchedule::new(runtime.clone());
921
7
        let task_schedule = Mutex::new(Some(task_schedule));
922

            
923
        // We load the cached protocol recommendations unconditionally: the caller needs them even
924
        // if it does not try to load the reset of the cache.
925
7
        let protocols = {
926
7
            let store = store.store.lock().expect("lock poisoned");
927
7
            store
928
7
                .cached_protocol_recommendations()?
929
7
                .map(|(t, p)| (t, Arc::new(p)))
930
        };
931

            
932
7
        Ok(DirMgr {
933
7
            config: config.into(),
934
7
            store: store.store,
935
7
            netdir,
936
7
            protocols: Mutex::new(protocols),
937
7
            default_parameters,
938
7
            events,
939
7
            send_status,
940
7
            receive_status,
941
7
            circmgr,
942
7
            runtime,
943
7
            offline,
944
7
            bootstrap_started: AtomicBool::new(false),
945
7
            #[cfg(feature = "dirfilter")]
946
7
            filter,
947
7
            task_schedule,
948
7
            task_handle,
949
7
        })
950
7
    }
951

            
952
    /// Load the latest non-pending non-expired directory from the
953
    /// cache, if it is newer than the one we have.
954
    ///
955
    /// Return false if there is no such consensus.
956
    fn load_directory(self: &Arc<Self>, attempt_id: AttemptId) -> Result<bool> {
957
        let state = state::GetConsensusState::new(
958
            self.runtime.clone(),
959
            self.config.get(),
960
            CacheUsage::CacheOnly,
961
            None,
962
            #[cfg(feature = "dirfilter")]
963
            self.filter
964
                .clone()
965
                .unwrap_or_else(|| Arc::new(crate::filter::NilFilter)),
966
        );
967
        let _ = bootstrap::load(self, Box::new(state), attempt_id)?;
968

            
969
        Ok(self.netdir.get().is_some())
970
    }
971

            
972
    /// Return a new asynchronous stream that will receive notification
973
    /// whenever the consensus has changed.
974
    ///
975
    /// Multiple events may be batched up into a single item: each time
976
    /// this stream yields an event, all you can assume is that the event has
977
    /// occurred at least once.
978
    pub fn events(&self) -> impl futures::Stream<Item = DirEvent> + use<R> {
979
        self.events.subscribe()
980
    }
981

            
982
    /// Try to load the text of a single document described by `doc` from
983
    /// storage.
984
3
    pub fn text(&self, doc: &DocId) -> Result<Option<DocumentText>> {
985
        use itertools::Itertools;
986
3
        let mut result = HashMap::new();
987
3
        let query: DocQuery = (*doc).into();
988
3
        let store = self.store.lock().expect("store lock poisoned");
989
3
        query.load_from_store_into(&mut result, &**store)?;
990
3
        let item = result.into_iter().at_most_one().map_err(|_| {
991
            Error::CacheCorruption("Found more than one entry in storage for given docid")
992
        })?;
993
3
        if let Some((docid, doctext)) = item {
994
2
            if &docid != doc {
995
                return Err(Error::CacheCorruption(
996
                    "Item from storage had incorrect docid.",
997
                ));
998
2
            }
999
2
            Ok(Some(doctext))
        } else {
1
            Ok(None)
        }
3
    }
    /// Load the text for a collection of documents.
    ///
    /// If many of the documents have the same type, this can be more
    /// efficient than calling [`text`](Self::text).
1
    pub fn texts<T>(&self, docs: T) -> Result<HashMap<DocId, DocumentText>>
1
    where
1
        T: IntoIterator<Item = DocId>,
    {
1
        let partitioned = docid::partition_by_type(docs);
1
        let mut result = HashMap::new();
1
        let store = self.store.lock().expect("store lock poisoned");
3
        for (_, query) in partitioned.into_iter() {
3
            query.load_from_store_into(&mut result, &**store)?;
        }
1
        Ok(result)
1
    }
    /// Given a request we sent and the response we got from a
    /// directory server, see whether we should expand that response
    /// into "something larger".
    ///
    /// Currently, this handles expanding consensus diffs, and nothing
    /// else.  We do it at this stage of our downloading operation
    /// because it requires access to the store.
6
    fn expand_response_text(&self, req: &ClientRequest, text: String) -> Result<String> {
6
        if let ClientRequest::Consensus(req) = req {
4
            if tor_consdiff::looks_like_diff(&text) {
2
                if let Some(old_d) = req.old_consensus_digests().next() {
2
                    let db_val = {
2
                        let s = self.store.lock().expect("Directory storage lock poisoned");
2
                        s.consensus_by_sha3_digest_of_signed_part(old_d)?
                    };
2
                    if let Some((old_consensus, meta)) = db_val {
2
                        info!("Applying a consensus diff");
2
                        let new_consensus = tor_consdiff::apply_diff(
2
                            old_consensus.as_str()?,
2
                            &text,
2
                            Some(*meta.sha3_256_of_signed()),
2
                            tor_consdiff::DiffSizeStrictness::Apply,
                        )?;
2
                        new_consensus.check_digest()?;
1
                        return Ok(new_consensus.to_string());
                    }
                }
                return Err(Error::Unwanted(
                    "Received a consensus diff we did not ask for",
                ));
2
            }
2
        }
4
        Ok(text)
6
    }
    /// If `state` has netdir changes to apply, apply them to our netdir.
8
    fn apply_netdir_changes(
8
        self: &Arc<Self>,
8
        state: &mut Box<dyn DirState>,
8
        store: &mut dyn Store,
8
    ) -> Result<()> {
8
        if let Some(change) = state.get_netdir_change() {
            match change {
                NetDirChange::AttemptReplace {
                    netdir,
                    consensus_meta,
                } => {
                    // Check the new netdir is sufficient, if we have a circmgr.
                    // (Unwraps are fine because the `Option` is `Some` until we take it.)
                    if let Some(ref cm) = self.circmgr {
                        if !cm
                            .netdir_is_sufficient(netdir.as_ref().expect("AttemptReplace had None"))
                        {
                            debug!("Got a new NetDir, but it doesn't have enough guards yet.");
                            return Ok(());
                        }
                    }
                    let is_stale = {
                        // Done inside a block to not hold a long-lived copy of the NetDir.
                        self.netdir
                            .get()
                            .map(|x| {
                                x.lifetime().valid_after()
                                    > netdir
                                        .as_ref()
                                        .expect("AttemptReplace had None")
                                        .lifetime()
                                        .valid_after()
                            })
                            .unwrap_or(false)
                    };
                    if is_stale {
                        warn!("Got a new NetDir, but it's older than the one we currently have!");
                        return Err(Error::NetDirOlder);
                    }
                    let cfg = self.config.get();
                    let mut netdir = netdir.take().expect("AttemptReplace had None");
                    netdir.replace_overridden_parameters(&cfg.override_net_params);
                    self.netdir.replace(netdir);
                    self.events.publish(DirEvent::NewConsensus);
                    self.events.publish(DirEvent::NewDescriptors);
                    info!("Marked consensus usable.");
                    if !store.is_readonly() {
                        store.mark_consensus_usable(consensus_meta)?;
                        // Now that a consensus is usable, older consensuses may
                        // need to expire.
                        store.expire_all(&crate::storage::EXPIRATION_DEFAULTS)?;
                    }
                    Ok(())
                }
                NetDirChange::AddMicrodescs(mds) => {
                    self.netdir.mutate(|netdir| {
                        for md in mds.drain(..) {
                            netdir.add_microdesc(md);
                        }
                        Ok(())
                    })?;
                    self.events.publish(DirEvent::NewDescriptors);
                    Ok(())
                }
                NetDirChange::SetRequiredProtocol { timestamp, protos } => {
                    if !store.is_readonly() {
                        store.update_protocol_recommendations(timestamp, protos.as_ref())?;
                    }
                    let mut pr = self.protocols.lock().expect("Poisoned lock");
                    *pr = Some((timestamp, protos));
                    self.events.publish(DirEvent::NewProtocolRecommendation);
                    Ok(())
                }
            }
        } else {
8
            Ok(())
        }
8
    }
    /// Experimental; temporary: Return a directory plugin to be used while tor-dirserver is a work
    /// in progress.
    #[cfg(feature = "dir-plugin")]
    pub fn get_plugin(&self) -> as_plugin::DirPlugin {
        as_plugin::DirPlugin {
            store: Arc::clone(&self.store),
        }
    }
}
/// A degree of readiness for a given directory state object.
#[derive(Debug, Copy, Clone)]
enum Readiness {
    /// There is no more information to download.
    Complete,
    /// There is more information to download, but we don't need to
    Usable,
}
/// Try to upgrade a weak reference to a DirMgr, and give an error on
/// failure.
12
fn upgrade_weak_ref<T>(weak: &Weak<T>) -> Result<Arc<T>> {
12
    Weak::upgrade(weak).ok_or(Error::ManagerDropped)
12
}
/// Given a time `now`, and an amount of tolerated clock skew `tolerance`,
/// return the age of the oldest consensus that we should request at that time.
4
pub(crate) fn default_consensus_cutoff(
4
    now: SystemTime,
4
    tolerance: &DirTolerance,
4
) -> Result<SystemTime> {
    /// We _always_ allow at least this much age in our consensuses, to account
    /// for the fact that consensuses have some lifetime.
    const MIN_AGE_TO_ALLOW: Duration = Duration::from_secs(3 * 3600);
4
    let allow_skew = std::cmp::max(MIN_AGE_TO_ALLOW, tolerance.post_valid_tolerance());
4
    let cutoff = time::OffsetDateTime::from(now - allow_skew);
    // We now round cutoff to the next hour, so that we aren't leaking our exact
    // time to the directory cache.
    //
    // With the time crate, it's easier to calculate the "next hour" by rounding
    // _down_ then adding an hour; rounding up would sometimes require changing
    // the date too.
4
    let (h, _m, _s) = cutoff.to_hms();
4
    let cutoff = cutoff.replace_time(
4
        time::Time::from_hms(h, 0, 0)
4
            .map_err(tor_error::into_internal!("Failed clock calculation"))?,
    );
4
    let cutoff = cutoff + Duration::from_secs(3600);
4
    Ok(cutoff.into())
4
}
/// Return a list of the protocols [supported](tor_protover::doc_supported) by this crate
/// when running as a client.
36
pub fn supported_client_protocols() -> tor_protover::Protocols {
    use tor_protover::named::*;
    // WARNING: REMOVING ELEMENTS FROM THIS LIST CAN BE DANGEROUS!
    // SEE [`tor_protover::doc_changing`]
36
    [
36
        //
36
        DIRCACHE_CONSDIFF,
36
    ]
36
    .into_iter()
36
    .collect()
36
}
#[cfg(test)]
mod test {
    // @@ begin test lint list maintained by maint/add_warning @@
    #![allow(clippy::bool_assert_comparison)]
    #![allow(clippy::clone_on_copy)]
    #![allow(clippy::dbg_macro)]
    #![allow(clippy::mixed_attributes_style)]
    #![allow(clippy::print_stderr)]
    #![allow(clippy::print_stdout)]
    #![allow(clippy::single_char_pattern)]
    #![allow(clippy::unwrap_used)]
    #![allow(clippy::unchecked_time_subtraction)]
    #![allow(clippy::useless_vec)]
    #![allow(clippy::needless_pass_by_value)]
    #![allow(clippy::string_slice)] // See arti#2571
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
    use super::*;
    use crate::docmeta::{AuthCertMeta, ConsensusMeta};
    use std::time::Duration;
    use tempfile::TempDir;
    use tor_basic_utils::test_rng::testing_rng;
    use tor_netdoc::doc::netstatus::ConsensusFlavor;
    use tor_netdoc::doc::{authcert::AuthCertKeyIds, netstatus::Lifetime};
    use tor_rtcompat::SleepProvider;
    #[test]
    fn protocols() {
        let pr = supported_client_protocols();
        let expected = "DirCache=2".parse().unwrap();
        assert_eq!(pr, expected);
    }
    pub(crate) fn new_mgr<R: Runtime>(runtime: R) -> (TempDir, DirMgr<R>) {
        let dir = TempDir::new().unwrap();
        let config = DirMgrConfig {
            cache_dir: dir.path().into(),
            ..Default::default()
        };
        let store = DirMgrStore::new(&config, runtime.clone(), false).unwrap();
        let dirmgr = DirMgr::from_config(config, runtime, store, None, false).unwrap();
        (dir, dirmgr)
    }
    #[test]
    fn failing_accessors() {
        tor_rtcompat::test_with_one_runtime!(|rt| async {
            let (_tempdir, mgr) = new_mgr(rt);
            assert!(mgr.circmgr().is_err());
            assert!(mgr.netdir(Timeliness::Unchecked).is_err());
        });
    }
    #[test]
    fn load_and_store_internals() {
        tor_rtcompat::test_with_one_runtime!(|rt| async {
            let now = rt.wallclock();
            let tomorrow = now + Duration::from_secs(86400);
            let later = tomorrow + Duration::from_secs(86400);
            let (_tempdir, mgr) = new_mgr(rt);
            // Seed the storage with a bunch of junk.
            let d1 = [5_u8; 32];
            let d2 = [7; 32];
            let d3 = [42; 32];
            let d4 = [99; 20];
            let d5 = [12; 20];
            let certid1 = AuthCertKeyIds {
                id_fingerprint: d4.into(),
                sk_fingerprint: d5.into(),
            };
            let certid2 = AuthCertKeyIds {
                id_fingerprint: d5.into(),
                sk_fingerprint: d4.into(),
            };
            {
                let mut store = mgr.store.lock().unwrap();
                store
                    .store_microdescs(
                        &[
                            ("Fake micro 1", &d1),
                            ("Fake micro 2", &d2),
                            ("Fake micro 3", &d3),
                        ],
                        now,
                    )
                    .unwrap();
                #[cfg(feature = "routerdesc")]
                store
                    .store_routerdescs(&[("Fake rd1", now, &d4), ("Fake rd2", now, &d5)])
                    .unwrap();
                store
                    .store_authcerts(&[
                        (
                            AuthCertMeta::new(certid1, now, tomorrow),
                            "Fake certificate one",
                        ),
                        (
                            AuthCertMeta::new(certid2, now, tomorrow),
                            "Fake certificate two",
                        ),
                    ])
                    .unwrap();
                let cmeta = ConsensusMeta::new(
                    Lifetime::new(now, tomorrow, later).unwrap(),
                    [102; 32],
                    [103; 32],
                );
                store
                    .store_consensus(&cmeta, ConsensusFlavor::Microdesc, false, "Fake consensus!")
                    .unwrap();
            }
            // Try to get it with text().
            let t1 = mgr.text(&DocId::Microdesc(d1)).unwrap().unwrap();
            assert_eq!(t1.as_str(), Ok("Fake micro 1"));
            let t2 = mgr
                .text(&DocId::LatestConsensus {
                    flavor: ConsensusFlavor::Microdesc,
                    cache_usage: CacheUsage::CacheOkay,
                })
                .unwrap()
                .unwrap();
            assert_eq!(t2.as_str(), Ok("Fake consensus!"));
            let t3 = mgr.text(&DocId::Microdesc([255; 32])).unwrap();
            assert!(t3.is_none());
            // Now try texts()
            let d_bogus = DocId::Microdesc([255; 32]);
            let res = mgr
                .texts(vec![
                    DocId::Microdesc(d2),
                    DocId::Microdesc(d3),
                    d_bogus,
                    DocId::AuthCert(certid2),
                    #[cfg(feature = "routerdesc")]
                    DocId::RouterDesc(d5),
                ])
                .unwrap();
            assert_eq!(
                res.get(&DocId::Microdesc(d2)).unwrap().as_str(),
                Ok("Fake micro 2")
            );
            assert_eq!(
                res.get(&DocId::Microdesc(d3)).unwrap().as_str(),
                Ok("Fake micro 3")
            );
            assert!(!res.contains_key(&d_bogus));
            assert_eq!(
                res.get(&DocId::AuthCert(certid2)).unwrap().as_str(),
                Ok("Fake certificate two")
            );
            #[cfg(feature = "routerdesc")]
            assert_eq!(
                res.get(&DocId::RouterDesc(d5)).unwrap().as_str(),
                Ok("Fake rd2")
            );
        });
    }
    #[test]
    fn make_consensus_request() {
        tor_rtcompat::test_with_one_runtime!(|rt| async {
            let now = rt.wallclock();
            let tomorrow = now + Duration::from_secs(86400);
            let later = tomorrow + Duration::from_secs(86400);
            let (_tempdir, mgr) = new_mgr(rt);
            let config = DirMgrConfig::default();
            // Try with an empty store.
            let req = {
                let store = mgr.store.lock().unwrap();
                bootstrap::make_consensus_request(
                    now,
                    ConsensusFlavor::Microdesc,
                    &**store,
                    &config,
                )
                .unwrap()
            };
            let tolerance = DirTolerance::default().post_valid_tolerance();
            match req {
                ClientRequest::Consensus(r) => {
                    assert_eq!(r.old_consensus_digests().count(), 0);
                    let date = r.last_consensus_date().unwrap();
                    assert!(date >= now - tolerance);
                    assert!(date <= now - tolerance + Duration::from_secs(3600));
                }
                _ => panic!("Wrong request type"),
            }
            // Add a fake consensus record.
            let d_prev = [42; 32];
            {
                let mut store = mgr.store.lock().unwrap();
                let cmeta = ConsensusMeta::new(
                    Lifetime::new(now, tomorrow, later).unwrap(),
                    d_prev,
                    [103; 32],
                );
                store
                    .store_consensus(&cmeta, ConsensusFlavor::Microdesc, false, "Fake consensus!")
                    .unwrap();
            }
            // Now try again.
            let req = {
                let store = mgr.store.lock().unwrap();
                bootstrap::make_consensus_request(
                    now,
                    ConsensusFlavor::Microdesc,
                    &**store,
                    &config,
                )
                .unwrap()
            };
            match req {
                ClientRequest::Consensus(r) => {
                    let ds: Vec<_> = r.old_consensus_digests().collect();
                    assert_eq!(ds.len(), 1);
                    assert_eq!(ds[0], &d_prev);
                    assert_eq!(r.last_consensus_date(), Some(now));
                }
                _ => panic!("Wrong request type"),
            }
        });
    }
    #[test]
    fn make_other_requests() {
        tor_rtcompat::test_with_one_runtime!(|rt| async {
            use rand::RngExt;
            let (_tempdir, mgr) = new_mgr(rt);
            let certid1 = AuthCertKeyIds {
                id_fingerprint: [99; 20].into(),
                sk_fingerprint: [100; 20].into(),
            };
            let mut rng = testing_rng();
            #[cfg(feature = "routerdesc")]
            let rd_ids: Vec<DocId> = (0..1000).map(|_| DocId::RouterDesc(rng.random())).collect();
            let md_ids: Vec<DocId> = (0..1000).map(|_| DocId::Microdesc(rng.random())).collect();
            let config = DirMgrConfig::default();
            // Try an authcert.
            let query = DocId::AuthCert(certid1);
            let store = mgr.store.lock().unwrap();
            let reqs =
                bootstrap::make_requests_for_documents(&mgr.runtime, &[query], &**store, &config)
                    .unwrap();
            assert_eq!(reqs.len(), 1);
            let req = &reqs[0];
            if let ClientRequest::AuthCert(r) = req {
                assert_eq!(r.keys().next(), Some(&certid1));
            } else {
                panic!();
            }
            // Try a bunch of mds.
            let reqs =
                bootstrap::make_requests_for_documents(&mgr.runtime, &md_ids, &**store, &config)
                    .unwrap();
            assert_eq!(reqs.len(), 2);
            assert!(matches!(reqs[0], ClientRequest::Microdescs(_)));
            // Try a bunch of rds.
            #[cfg(feature = "routerdesc")]
            {
                let reqs = bootstrap::make_requests_for_documents(
                    &mgr.runtime,
                    &rd_ids,
                    &**store,
                    &config,
                )
                .unwrap();
                assert_eq!(reqs.len(), 2);
                assert!(matches!(reqs[0], ClientRequest::RouterDescs(_)));
            }
        });
    }
    #[test]
    fn expand_response() {
        tor_rtcompat::test_with_one_runtime!(|rt| async {
            let now = rt.wallclock();
            let day = Duration::from_secs(86400);
            let config = DirMgrConfig::default();
            let (_tempdir, mgr) = new_mgr(rt);
            // Try a simple request: nothing should happen.
            let q = DocId::Microdesc([99; 32]);
            let r = {
                let store = mgr.store.lock().unwrap();
                bootstrap::make_requests_for_documents(&mgr.runtime, &[q], &**store, &config)
                    .unwrap()
            };
            let expanded = mgr.expand_response_text(&r[0], "ABC".to_string());
            assert_eq!(&expanded.unwrap(), "ABC");
            // Try a consensus response that doesn't look like a diff in
            // response to a query that doesn't ask for one.
            let latest_id = DocId::LatestConsensus {
                flavor: ConsensusFlavor::Microdesc,
                cache_usage: CacheUsage::CacheOkay,
            };
            let r = {
                let store = mgr.store.lock().unwrap();
                bootstrap::make_requests_for_documents(
                    &mgr.runtime,
                    &[latest_id],
                    &**store,
                    &config,
                )
                .unwrap()
            };
            let expanded = mgr.expand_response_text(&r[0], "DEF".to_string());
            assert_eq!(&expanded.unwrap(), "DEF");
            // Now stick some metadata and a string into the storage so that
            // we can ask for a diff.
            {
                let mut store = mgr.store.lock().unwrap();
                let d_in = [0x99; 32]; // This one, we can fake.
                let cmeta = ConsensusMeta::new(
                    Lifetime::new(now, now + day, now + 2 * day).unwrap(),
                    d_in,
                    d_in,
                );
                store
                    .store_consensus(
                        &cmeta,
                        ConsensusFlavor::Microdesc,
                        false,
                        "line 1\nline2\nline 3\n",
                    )
                    .unwrap();
            }
            // Try expanding something that isn't a consensus, even if we'd like
            // one.
            let r = {
                let store = mgr.store.lock().unwrap();
                bootstrap::make_requests_for_documents(
                    &mgr.runtime,
                    &[latest_id],
                    &**store,
                    &config,
                )
                .unwrap()
            };
            let expanded = mgr.expand_response_text(&r[0], "hello".to_string());
            assert_eq!(&expanded.unwrap(), "hello");
            // Finally, try "expanding" a diff (by applying it and checking the digest.
            let diff = "network-status-diff-version 1
hash 9999999999999999999999999999999999999999999999999999999999999999 8382374ca766873eb0d2530643191c6eaa2c5e04afa554cbac349b5d0592d300
2c
replacement line
.
".to_string();
            let expanded = mgr.expand_response_text(&r[0], diff);
            assert_eq!(expanded.unwrap(), "line 1\nreplacement line\nline 3\n");
            // If the digest is wrong, that should get rejected.
            let diff = "network-status-diff-version 1
hash 9999999999999999999999999999999999999999999999999999999999999999 9999999999999999999999999999999999999999999999999999999999999999
2c
replacement line
.
".to_string();
            let expanded = mgr.expand_response_text(&r[0], diff);
            assert!(expanded.is_err());
        });
    }
}