1
//! The onion service publisher reactor.
2
//!
3
//! Generates and publishes hidden service descriptors in response to various events.
4
//!
5
//! [`Reactor::run`] is the entry-point of the reactor. It starts the reactor,
6
//! and runs until [`Reactor::run_once`] returns [`ShutdownStatus::Terminate`]
7
//! or a fatal error occurs. `ShutdownStatus::Terminate` is returned if
8
//! any of the channels the reactor is receiving events from is closed
9
//! (i.e. when the senders are dropped).
10
//!
11
//! ## Publisher status
12
//!
13
//! The publisher has an internal [`PublishStatus`], distinct from its [`State`],
14
//! which is used for onion service status reporting.
15
//!
16
//! The main loop of the reactor reads the current `PublishStatus` from `publish_status_rx`,
17
//! and responds by generating and publishing a new descriptor if needed.
18
//!
19
//! See [`PublishStatus`] and [`Reactor::publish_status_rx`] for more details.
20
//!
21
//! ## When do we publish?
22
//!
23
//! We generate and publish a new descriptor if
24
//!   * the introduction points have changed
25
//!   * the onion service configuration has changed in a meaningful way (for example,
26
//!     if the `restricted_discovery` configuration or its [`Anonymity`](crate::Anonymity)
27
//!     has changed. See [`OnionServiceConfigPublisherView`]).
28
//!   * there is a new consensus
29
//!   * it is time to republish the descriptor (after we upload a descriptor,
30
//!     we schedule it for republishing at a random time between 60 minutes and 120 minutes
31
//!     in the future)
32
//!
33
//! ## Onion service status
34
//!
35
//! With respect to [`OnionServiceStatus`] reporting,
36
//! the following state transitions are possible:
37
//!
38
//!
39
//! ```ignore
40
//!
41
//!                 update_publish_status(UploadScheduled|AwaitingIpts|RateLimited)
42
//!                +---------------------------------------+
43
//!                |                                       |
44
//!                |                                       v
45
//!                |                               +---------------+
46
//!                |                               | Bootstrapping |
47
//!                |                               +---------------+
48
//!                |                                       |
49
//!                |                                       |           uploaded to at least
50
//!                |  not enough HsDir uploads succeeded   |        some HsDirs from each ring
51
//!                |         +-----------------------------+-----------------------+
52
//!                |         |                             |                       |
53
//!                |         |              all HsDir uploads succeeded            |
54
//!                |         |                             |                       |
55
//!                |         v                             v                       v
56
//!                |  +---------------------+         +---------+        +---------------------+
57
//!                |  | DegradedUnreachable |         | Running |        |  DegradedReachable  |
58
//! +----------+   |  +---------------------+         +---------+        +---------------------+
59
//! | Shutdown |-- |         |                           |                        |
60
//! +----------+   |         |                           |                        |
61
//!                |         |                           |                        |
62
//!                |         |                           |                        |
63
//!                |         +---------------------------+------------------------+
64
//!                |                                     |   invalid authorized_clients
65
//!                |                                     |      after handling config change
66
//!                |                                     |
67
//!                |                                     v
68
//!                |     run_once() returns an error +--------+
69
//!                +-------------------------------->| Broken |
70
//!                                                  +--------+
71
//! ```
72
//!
73
//! We can also transition from `Broken`, `DegradedReachable`, or `DegradedUnreachable`
74
//! back to `Bootstrapping` (those transitions were omitted for brevity).
75

            
76
use tor_circmgr::ServiceOnionServiceDirTunnel;
77
use tor_config::file_watcher::{
78
    self, Event as FileEvent, FileEventReceiver, FileEventSender, FileWatcher, FileWatcherBuilder,
79
};
80
use tor_config_path::{CfgPath, CfgPathResolver};
81
use tor_dirclient::{RequestError, SourceInfo};
82
use tor_netdir::{DirEvent, NetDir};
83
use tor_persist::state_dir::InstanceRawSubdir;
84
use tracing::instrument;
85

            
86
use crate::config::OnionServiceConfigPublisherView;
87
use crate::config::restricted_discovery::{
88
    DirectoryKeyProviderList, RestrictedDiscoveryConfig, RestrictedDiscoveryKeys,
89
};
90
use crate::status::{DescUploadRetryError, Problem};
91

            
92
use super::*;
93
use derive_more::From;
94
use fs_mistrust::anon_home::PathExt;
95

            
96
// TODO-CLIENT-AUTH: perhaps we should add a separate CONFIG_CHANGE_REPUBLISH_DEBOUNCE_INTERVAL
97
// for rate-limiting the publish jobs triggered by a change in the config?
98
//
99
// Currently the descriptor publish tasks triggered by changes in the config
100
// are rate-limited via the usual rate limiting mechanism
101
// (which rate-limits the uploads for 1m).
102
//
103
// I think this is OK for now, but we might need to rethink this if it becomes problematic
104
// (for example, we might want an even longer rate-limit, or to reset any existing rate-limits
105
// each time the config is modified).
106

            
107
/// The upload rate-limiting threshold.
108
///
109
/// Before initiating an upload, the reactor checks if the last upload was at least
110
/// `UPLOAD_RATE_LIM_THRESHOLD` seconds ago. If so, it uploads the descriptor to all HsDirs that
111
/// need it. If not, it schedules the upload to happen `UPLOAD_RATE_LIM_THRESHOLD` seconds from the
112
/// current time.
113
//
114
// TODO: We may someday need to tune this value; it was chosen more or less arbitrarily.
115
const UPLOAD_RATE_LIM_THRESHOLD: Duration = Duration::from_secs(60);
116

            
117
/// The maximum number of concurrent upload tasks per time period.
118
//
119
// TODO: this value was arbitrarily chosen and may not be optimal.  For now, it
120
// will have no effect, since the current number of replicas is far less than
121
// this value.
122
//
123
// The uploads for all TPs happen in parallel.  As a result, the actual limit for the maximum
124
// number of concurrent upload tasks is multiplied by a number which depends on the TP parameters
125
// (currently 2, which means the concurrency limit will, in fact, be 32).
126
//
127
// We should try to decouple this value from the TP parameters.
128
const MAX_CONCURRENT_UPLOADS: usize = 16;
129

            
130
/// The maximum time allowed for uploading a descriptor to a single HSDir,
131
/// across all attempts.
132
pub(crate) const OVERALL_UPLOAD_TIMEOUT: Duration = Duration::from_secs(5 * 60);
133

            
134
/// A reactor for the HsDir [`Publisher`]
135
///
136
/// The entrypoint is [`Reactor::run`].
137
#[must_use = "If you don't call run() on the reactor, it won't publish any descriptors."]
138
pub(super) struct Reactor<R: Runtime, M: Mockable> {
139
    /// The immutable, shared inner state.
140
    imm: Arc<Immutable<R, M>>,
141
    /// A source for new network directories that we use to determine
142
    /// our HsDirs.
143
    dir_provider: Arc<dyn NetDirProvider>,
144
    /// The mutable inner state,
145
    inner: Arc<Mutex<Inner>>,
146
    /// A channel for receiving IPT change notifications.
147
    ipt_watcher: IptsPublisherView,
148
    /// A channel for receiving onion service config change notifications.
149
    config_rx: watch::Receiver<Arc<OnionServiceConfig>>,
150
    /// A channel for receiving restricted discovery key_dirs change notifications.
151
    key_dirs_rx: FileEventReceiver,
152
    /// A channel for sending restricted discovery key_dirs change notifications.
153
    ///
154
    /// A copy of this sender is handed out to every `FileWatcher` created.
155
    key_dirs_tx: FileEventSender,
156
    /// A channel for receiving updates regarding our [`PublishStatus`].
157
    ///
158
    /// The main loop of the reactor watches for updates on this channel.
159
    ///
160
    /// When the [`PublishStatus`] changes to [`UploadScheduled`](PublishStatus::UploadScheduled),
161
    /// we can start publishing descriptors.
162
    ///
163
    /// If the [`PublishStatus`] is [`AwaitingIpts`](PublishStatus::AwaitingIpts), publishing is
164
    /// paused until we receive a notification on `ipt_watcher` telling us the IPT manager has
165
    /// established some introduction points.
166
    publish_status_rx: watch::Receiver<PublishStatus>,
167
    /// A sender for updating our [`PublishStatus`].
168
    ///
169
    /// When our [`PublishStatus`] changes to [`UploadScheduled`](PublishStatus::UploadScheduled),
170
    /// we can start publishing descriptors.
171
    publish_status_tx: watch::Sender<PublishStatus>,
172
    /// A channel for sending upload completion notifications.
173
    ///
174
    /// This channel is polled in the main loop of the reactor.
175
    upload_task_complete_rx: mpsc::Receiver<TimePeriodUploadResult>,
176
    /// A channel for receiving upload completion notifications.
177
    ///
178
    /// A copy of this sender is handed to each upload task.
179
    upload_task_complete_tx: mpsc::Sender<TimePeriodUploadResult>,
180
    /// A sender for notifying any pending upload tasks that the reactor is shutting down.
181
    ///
182
    /// Receivers can use this channel to find out when reactor is dropped.
183
    ///
184
    /// This is currently only used in [`upload_for_time_period`](Reactor::upload_for_time_period).
185
    /// Any future background tasks can also use this channel to detect if the reactor is dropped.
186
    ///
187
    /// Closing this channel will cause any pending upload tasks to be dropped.
188
    shutdown_tx: broadcast::Sender<Void>,
189
    /// Path resolver for configuration files.
190
    path_resolver: Arc<CfgPathResolver>,
191
    /// Queue on which we receive messages from the [`PowManager`] telling us that a seed has
192
    /// rotated and thus we need to republish the descriptor for a particular time period.
193
    update_from_pow_manager_rx: mpsc::Receiver<TimePeriod>,
194
}
195

            
196
/// The immutable, shared state of the descriptor publisher reactor.
197
#[derive(Clone)]
198
struct Immutable<R: Runtime, M: Mockable> {
199
    /// The runtime.
200
    runtime: R,
201
    /// Mockable state.
202
    ///
203
    /// This is used for launching circuits and for obtaining random number generators.
204
    mockable: M,
205
    /// The service for which we're publishing descriptors.
206
    nickname: HsNickname,
207
    /// The key manager,
208
    keymgr: Arc<KeyMgr>,
209
    /// A sender for updating the status of the onion service.
210
    status_tx: PublisherStatusSender,
211
    /// Proof-of-work state.
212
    pow_manager: Arc<PowManager<R>>,
213
    /// The directory to dump invalid HS descriptors to.
214
    bad_hsdescs_dir: InstanceRawSubdir,
215
}
216

            
217
impl<R: Runtime, M: Mockable> Immutable<R, M> {
218
    /// Create an [`AesOpeKey`] for generating revision counters for the descriptors associated
219
    /// with the specified [`TimePeriod`].
220
    ///
221
    /// If the onion service is not running in offline mode, the key of the returned `AesOpeKey` is
222
    /// the private part of the blinded identity key. Otherwise, the key is the private part of the
223
    /// descriptor signing key.
224
    ///
225
    /// Returns an error if the service is running in offline mode and the descriptor signing
226
    /// keypair of the specified `period` is not available.
227
    //
228
    // TODO (#1194): we don't support "offline" mode (yet), so this always returns an AesOpeKey
229
    // built from the blinded id key
230
72
    fn create_ope_key(&self, period: TimePeriod) -> Result<AesOpeKey, FatalError> {
231
72
        let ope_key = match read_blind_id_keypair(&self.keymgr, &self.nickname, period)? {
232
72
            Some(key) => {
233
72
                let key: ed25519::ExpandedKeypair = key.into();
234
72
                key.to_secret_key_bytes()[0..32]
235
72
                    .try_into()
236
72
                    .expect("Wrong length on slice")
237
            }
238
            None => {
239
                // TODO (#1194): we don't support externally provisioned keys (yet), so this branch
240
                // is unreachable (for now).
241
                let desc_sign_key_spec =
242
                    DescSigningKeypairSpecifier::new(self.nickname.clone(), period);
243
                let key: ed25519::Keypair = self
244
                    .keymgr
245
                    .get::<HsDescSigningKeypair>(&desc_sign_key_spec)?
246
                    // TODO (#1194): internal! is not the right type for this error (we need an
247
                    // error type for the case where a hidden service running in offline mode has
248
                    // run out of its pre-previsioned keys).
249
                    //
250
                    // This will be addressed when we add support for offline hs_id mode
251
                    .ok_or_else(|| {
252
                        internal!(
253
                            "identity keys are offline, but descriptor signing key is unavailable?!"
254
                        )
255
                    })?
256
                    .into();
257
                key.to_bytes()
258
            }
259
        };
260

            
261
72
        Ok(AesOpeKey::from_secret(&ope_key))
262
72
    }
263

            
264
    /// Generate a revision counter for a descriptor associated with the specified
265
    /// [`TimePeriod`].
266
    ///
267
    /// Returns a revision counter generated according to the [encrypted time in period] scheme.
268
    ///
269
    /// [encrypted time in period]: https://spec.torproject.org/rend-spec/revision-counter-mgt.html#encrypted-time
270
72
    fn generate_revision_counter(
271
72
        &self,
272
72
        params: &HsDirParams,
273
72
        now: SystemTime,
274
72
    ) -> Result<RevisionCounter, FatalError> {
275
        // TODO: in the future, we might want to compute ope_key once per time period (as oppposed
276
        // to each time we generate a new descriptor), for performance reasons.
277
72
        let ope_key = self.create_ope_key(params.time_period())?;
278

            
279
        // TODO: perhaps this should be moved to a new HsDirParams::offset_within_sr() function
280
72
        let srv_start = params.start_of_shard_rand_period();
281
72
        let offset = params.offset_within_srv_period(now).ok_or_else(|| {
282
            internal!(
283
                "current wallclock time not within SRV range?! (now={:?}, SRV_start={:?})",
284
                now,
285
                srv_start
286
            )
287
        })?;
288
72
        let rev = ope_key.encrypt(offset);
289

            
290
72
        Ok(RevisionCounter::from(rev))
291
72
    }
292
}
293

            
294
/// Mockable state for the descriptor publisher reactor.
295
///
296
/// This enables us to mock parts of the [`Reactor`] for testing purposes.
297
#[async_trait]
298
pub(crate) trait Mockable: Clone + Send + Sync + Sized + 'static {
299
    /// The type of random number generator.
300
    type Rng: rand::Rng + rand::CryptoRng;
301

            
302
    /// The type of client circuit.
303
    type Tunnel: MockableDirTunnel;
304

            
305
    /// Return a random number generator.
306
    fn thread_rng(&self) -> Self::Rng;
307

            
308
    /// Create a circuit of the specified `kind` to `target`.
309
    async fn get_or_launch_hs_dir<T>(
310
        &self,
311
        netdir: &NetDir,
312
        target: T,
313
    ) -> Result<Self::Tunnel, tor_circmgr::Error>
314
    where
315
        T: CircTarget + Send + Sync;
316

            
317
    /// Return an estimate-based value for how long we should allow a single
318
    /// directory upload operation to complete.
319
    ///
320
    /// Includes circuit construction, stream opening, upload, and waiting for a
321
    /// response.
322
    fn estimate_upload_timeout(&self) -> Duration;
323
}
324

            
325
/// Mockable client circuit
326
#[async_trait]
327
pub(crate) trait MockableDirTunnel: Send + Sync {
328
    /// The data stream type.
329
    type DataStream: AsyncRead + AsyncWrite + Send + Unpin;
330

            
331
    /// Start a new stream to the last relay in the circuit, using
332
    /// a BEGIN_DIR cell.
333
    async fn begin_dir_stream(&self) -> Result<Self::DataStream, tor_circmgr::Error>;
334

            
335
    /// Try to get a SourceInfo for this circuit, for using it in a directory request.
336
    fn source_info(&self) -> tor_proto::Result<Option<SourceInfo>>;
337
}
338

            
339
#[async_trait]
340
impl MockableDirTunnel for ServiceOnionServiceDirTunnel {
341
    type DataStream = tor_proto::client::stream::DataStream;
342

            
343
    async fn begin_dir_stream(&self) -> Result<Self::DataStream, tor_circmgr::Error> {
344
        Self::begin_dir_stream(self).await
345
    }
346

            
347
    fn source_info(&self) -> tor_proto::Result<Option<SourceInfo>> {
348
        SourceInfo::from_tunnel(self)
349
    }
350
}
351

            
352
/// The real version of the mockable state of the reactor.
353
#[derive(Clone, From, Into)]
354
pub(crate) struct Real<R: Runtime>(Arc<HsCircPool<R>>);
355

            
356
#[async_trait]
357
impl<R: Runtime> Mockable for Real<R> {
358
    type Rng = rand::rngs::ThreadRng;
359
    type Tunnel = ServiceOnionServiceDirTunnel;
360

            
361
    fn thread_rng(&self) -> Self::Rng {
362
        rand::rng()
363
    }
364

            
365
    #[instrument(level = "trace", skip_all)]
366
    async fn get_or_launch_hs_dir<T>(
367
        &self,
368
        netdir: &NetDir,
369
        target: T,
370
    ) -> Result<Self::Tunnel, tor_circmgr::Error>
371
    where
372
        T: CircTarget + Send + Sync,
373
    {
374
        self.0.get_or_launch_svc_dir(netdir, target).await
375
    }
376

            
377
    fn estimate_upload_timeout(&self) -> Duration {
378
        use tor_circmgr::timeouts::Action;
379
        let est_build = self.0.estimate_timeout(&Action::BuildCircuit { length: 4 });
380
        let est_roundtrip = self.0.estimate_timeout(&Action::RoundTrip { length: 4 });
381
        // We assume that in the worst case we'll have to wait for an entire
382
        // circuit construction and two round-trips to the hsdir.
383
        let est_total = est_build + est_roundtrip * 2;
384
        // We always allow _at least_ this much time, in case our estimate is
385
        // ridiculously low.
386
        let min_timeout = Duration::from_secs(30);
387
        max(est_total, min_timeout)
388
    }
389
}
390

            
391
/// The mutable state of a [`Reactor`].
392
struct Inner {
393
    /// The onion service config.
394
    config: Arc<OnionServiceConfigPublisherView>,
395
    /// Watcher for key_dirs.
396
    ///
397
    /// Set to `None` if the reactor is not running, or if `watch_configuration` is false.
398
    ///
399
    /// The watcher is recreated whenever the `restricted_discovery.key_dirs` change.
400
    file_watcher: Option<FileWatcher>,
401
    /// The relevant time periods.
402
    ///
403
    /// This includes the current time period, as well as any other time periods we need to be
404
    /// publishing descriptors for.
405
    ///
406
    /// This is empty until we fetch our first netdir in [`Reactor::run`].
407
    time_periods: Vec<TimePeriodContext>,
408
    /// Our most up to date netdir.
409
    ///
410
    /// This is initialized in [`Reactor::run`].
411
    netdir: Option<Arc<NetDir>>,
412
    /// The timestamp of our last upload.
413
    ///
414
    /// This is the time when the last update was _initiated_ (rather than completed), to prevent
415
    /// the publisher from spawning multiple upload tasks at once in response to multiple external
416
    /// events happening in quick succession, such as the IPT manager sending multiple IPT change
417
    /// notifications in a short time frame (#1142), or an IPT change notification that's
418
    /// immediately followed by a consensus change. Starting two upload tasks at once is not only
419
    /// inefficient, but it also causes the publisher to generate two different descriptors with
420
    /// the same revision counter (the revision counter is derived from the current timestamp),
421
    /// which ultimately causes the slower upload task to fail (see #1142).
422
    ///
423
    /// Note: This is only used for deciding when to reschedule a rate-limited upload. It is _not_
424
    /// used for retrying failed uploads (these are handled internally by
425
    /// [`Reactor::upload_descriptor_with_retries`]).
426
    last_uploaded: Option<Instant>,
427
    /// A max-heap containing the time periods for which we need to reupload the descriptor.
428
    // TODO: we are currently reuploading more than nececessary.
429
    // Ideally, this shouldn't contain contain duplicate TimePeriods,
430
    // because we only need to retain the latest reupload time for each time period.
431
    //
432
    // Currently, if, for some reason, we upload the descriptor multiple times for the same TP,
433
    // we will end up with multiple ReuploadTimer entries for that TP,
434
    // each of which will (eventually) result in a reupload.
435
    //
436
    // TODO: maybe this should just be a HashMap<TimePeriod, Instant>
437
    //
438
    // See https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1971#note_2994950
439
    reupload_timers: BinaryHeap<ReuploadTimer>,
440
    /// The restricted discovery authorized clients.
441
    ///
442
    /// `None`, unless the service is running in restricted discovery mode.
443
    authorized_clients: Option<Arc<RestrictedDiscoveryKeys>>,
444
}
445

            
446
/// The part of the reactor state that changes with every time period.
447
struct TimePeriodContext {
448
    /// The HsDir params.
449
    params: HsDirParams,
450
    /// The HsDirs to use in this time period.
451
    ///
452
    // We keep a list of `RelayIds` because we can't store a `Relay<'_>` inside the reactor
453
    // (the lifetime of a relay is tied to the lifetime of its corresponding `NetDir`. To
454
    // store `Relay<'_>`s in the reactor, we'd need a way of atomically swapping out both the
455
    // `NetDir` and the cached relays, and to convince Rust what we're doing is sound)
456
    hs_dirs: Vec<(RelayIds, DescriptorStatus)>,
457
    /// The revision counter of the last successful upload, if any.
458
    last_successful: Option<RevisionCounter>,
459
    /// The outcome of the last upload, if any.
460
    upload_results: Vec<HsDirUploadStatus>,
461
}
462

            
463
impl TimePeriodContext {
464
    /// Create a new `TimePeriodContext`.
465
    ///
466
    /// Any of the specified `old_hsdirs` also present in the new list of HsDirs
467
    /// (returned by `NetDir::hs_dirs_upload`) will have their `DescriptorStatus` preserved.
468
4
    fn new<'r>(
469
4
        params: HsDirParams,
470
4
        blind_id: HsBlindId,
471
4
        netdir: &Arc<NetDir>,
472
4
        old_hsdirs: impl Iterator<Item = &'r (RelayIds, DescriptorStatus)>,
473
4
        old_upload_results: Vec<HsDirUploadStatus>,
474
4
    ) -> Result<Self, FatalError> {
475
4
        let period = params.time_period();
476
4
        let hs_dirs = Self::compute_hsdirs(period, blind_id, netdir, old_hsdirs)?;
477
4
        let upload_results = old_upload_results
478
4
            .into_iter()
479
4
            .filter(|res|
480
                // Check if the HsDir of this result still exists
481
                hs_dirs
482
                    .iter()
483
                    .any(|(relay_ids, _status)| relay_ids == &res.relay_ids))
484
4
            .collect();
485

            
486
4
        Ok(Self {
487
4
            params,
488
4
            hs_dirs,
489
4
            last_successful: None,
490
4
            upload_results,
491
4
        })
492
4
    }
493

            
494
    /// Recompute the HsDirs for this time period.
495
4
    fn compute_hsdirs<'r>(
496
4
        period: TimePeriod,
497
4
        blind_id: HsBlindId,
498
4
        netdir: &Arc<NetDir>,
499
4
        mut old_hsdirs: impl Iterator<Item = &'r (RelayIds, DescriptorStatus)>,
500
4
    ) -> Result<Vec<(RelayIds, DescriptorStatus)>, FatalError> {
501
4
        let hs_dirs = netdir.hs_dirs_upload(blind_id, period)?;
502

            
503
4
        Ok(hs_dirs
504
32
            .map(|hs_dir| {
505
32
                let mut builder = RelayIds::builder();
506
32
                if let Some(ed_id) = hs_dir.ed_identity() {
507
32
                    builder.ed_identity(*ed_id);
508
32
                }
509

            
510
32
                if let Some(rsa_id) = hs_dir.rsa_identity() {
511
32
                    builder.rsa_identity(*rsa_id);
512
32
                }
513

            
514
32
                let relay_id = builder.build().unwrap_or_else(|_| RelayIds::empty());
515

            
516
                // Have we uploaded the descriptor to thiw relay before? If so, we don't need to
517
                // reupload it unless it was already dirty and due for a reupload.
518
32
                let status = match old_hsdirs.find(|(id, _)| *id == relay_id) {
519
                    Some((_, status)) => *status,
520
32
                    None => DescriptorStatus::Dirty,
521
                };
522

            
523
32
                (relay_id, status)
524
32
            })
525
4
            .collect::<Vec<_>>())
526
4
    }
527

            
528
    /// Mark the descriptor dirty for all HSDirs of this time period.
529
9
    fn mark_all_dirty(&mut self) {
530
9
        self.hs_dirs
531
9
            .iter_mut()
532
72
            .for_each(|(_relay_id, status)| *status = DescriptorStatus::Dirty);
533
9
    }
534

            
535
    /// Update the upload result for this time period.
536
7
    fn set_upload_results(&mut self, upload_results: Vec<HsDirUploadStatus>) {
537
7
        self.upload_results = upload_results;
538
7
    }
539
}
540

            
541
/// An error that occurs while trying to upload a descriptor.
542
#[derive(Clone, Debug, thiserror::Error)]
543
#[non_exhaustive]
544
pub enum UploadError {
545
    /// An error that has occurred after we have contacted a directory cache and made a circuit to it.
546
    #[error("descriptor upload request failed: {}", _0.error)]
547
    Request(#[from] RequestFailedError),
548

            
549
    /// Failed to establish circuit to hidden service directory
550
    #[error("could not build circuit to HsDir")]
551
    Circuit(#[from] tor_circmgr::Error),
552

            
553
    /// Failed to establish stream to hidden service directory
554
    #[error("failed to establish directory stream to HsDir")]
555
    Stream(#[source] tor_circmgr::Error),
556

            
557
    /// An internal error.
558
    #[error("Internal error")]
559
    Bug(#[from] tor_error::Bug),
560
}
561
define_asref_dyn_std_error!(UploadError);
562

            
563
impl UploadError {
564
    /// Return true if this error is one that we should report as a suspicious event,
565
    /// along with the dirserver, and description of the relevant document.
566
32
    pub(crate) fn should_report_as_suspicious(&self) -> bool {
567
32
        match self {
568
32
            UploadError::Request(e) => e.error.should_report_as_suspicious_if_anon(),
569
            UploadError::Circuit(_) => false, // TODO prop360
570
            UploadError::Stream(_) => false,  // TODO prop360
571
            UploadError::Bug(_) => false,
572
        }
573
32
    }
574
}
575

            
576
impl<R: Runtime, M: Mockable> Reactor<R, M> {
577
    /// Create a new `Reactor`.
578
    #[allow(clippy::too_many_arguments)]
579
4
    pub(super) fn new(
580
4
        runtime: R,
581
4
        nickname: HsNickname,
582
4
        dir_provider: Arc<dyn NetDirProvider>,
583
4
        mockable: M,
584
4
        config: &OnionServiceConfig,
585
4
        ipt_watcher: IptsPublisherView,
586
4
        config_rx: watch::Receiver<Arc<OnionServiceConfig>>,
587
4
        status_tx: PublisherStatusSender,
588
4
        keymgr: Arc<KeyMgr>,
589
4
        path_resolver: Arc<CfgPathResolver>,
590
4
        pow_manager: Arc<PowManager<R>>,
591
4
        update_from_pow_manager_rx: mpsc::Receiver<TimePeriod>,
592
4
        bad_hsdescs_dir: InstanceRawSubdir,
593
4
    ) -> Self {
594
        /// The maximum size of the upload completion notifier channel.
595
        ///
596
        /// The channel we use this for is a futures::mpsc channel, which has a capacity of
597
        /// `UPLOAD_CHAN_BUF_SIZE + num-senders`. We don't need the buffer size to be non-zero, as
598
        /// each sender will send exactly one message.
599
        const UPLOAD_CHAN_BUF_SIZE: usize = 0;
600

            
601
        // Internally-generated instructions, no need for mq.
602
4
        let (upload_task_complete_tx, upload_task_complete_rx) =
603
4
            mpsc_channel_no_memquota(UPLOAD_CHAN_BUF_SIZE);
604

            
605
4
        let (publish_status_tx, publish_status_rx) = watch::channel();
606
        // Setting the buffer size to zero here is OK,
607
        // since we never actually send anything on this channel.
608
4
        let (shutdown_tx, _shutdown_rx) = broadcast::channel(0);
609

            
610
4
        let authorized_clients =
611
4
            Self::read_authorized_clients(&config.restricted_discovery, &path_resolver);
612

            
613
        // Create a channel for watching for changes in the configured
614
        // restricted_discovery.key_dirs.
615
4
        let (key_dirs_tx, key_dirs_rx) = file_watcher::channel();
616

            
617
4
        let imm = Immutable {
618
4
            runtime,
619
4
            mockable,
620
4
            nickname,
621
4
            keymgr,
622
4
            status_tx,
623
4
            pow_manager,
624
4
            bad_hsdescs_dir,
625
4
        };
626

            
627
4
        let inner = Inner {
628
4
            time_periods: vec![],
629
4
            config: Arc::new(config.into()),
630
4
            file_watcher: None,
631
4
            netdir: None,
632
4
            last_uploaded: None,
633
4
            reupload_timers: Default::default(),
634
4
            authorized_clients,
635
4
        };
636

            
637
4
        Self {
638
4
            imm: Arc::new(imm),
639
4
            inner: Arc::new(Mutex::new(inner)),
640
4
            dir_provider,
641
4
            ipt_watcher,
642
4
            config_rx,
643
4
            key_dirs_rx,
644
4
            key_dirs_tx,
645
4
            publish_status_rx,
646
4
            publish_status_tx,
647
4
            upload_task_complete_rx,
648
4
            upload_task_complete_tx,
649
4
            shutdown_tx,
650
4
            path_resolver,
651
4
            update_from_pow_manager_rx,
652
4
        }
653
4
    }
654

            
655
    /// Start the reactor.
656
    ///
657
    /// Under normal circumstances, this function runs indefinitely.
658
    ///
659
    /// Note: this also spawns the "reminder task" that we use to reschedule uploads whenever an
660
    /// upload fails or is rate-limited.
661
4
    pub(super) async fn run(mut self) -> Result<(), FatalError> {
662
4
        debug!(nickname=%self.imm.nickname, "starting descriptor publisher reactor");
663

            
664
        {
665
4
            let netdir = self
666
4
                .dir_provider
667
4
                .wait_for_netdir(Timeliness::Timely)
668
4
                .await?;
669
4
            let time_periods = self.compute_time_periods(&netdir, &[])?;
670

            
671
4
            let mut inner = self.inner.lock().expect("poisoned lock");
672

            
673
4
            inner.netdir = Some(netdir);
674
4
            inner.time_periods = time_periods;
675
        }
676

            
677
        // Create the initial key_dirs watcher.
678
4
        self.update_file_watcher();
679

            
680
        loop {
681
50
            match self.run_once().await {
682
46
                Ok(ShutdownStatus::Continue) => continue,
683
                Ok(ShutdownStatus::Terminate) => {
684
                    debug!(nickname=%self.imm.nickname, "descriptor publisher is shutting down!");
685

            
686
                    self.imm.status_tx.send_shutdown();
687
                    return Ok(());
688
                }
689
                Err(e) => {
690
                    error_report!(
691
                        e,
692
                        "HS service {}: descriptor publisher crashed!",
693
                        self.imm.nickname
694
                    );
695

            
696
                    self.imm.status_tx.send_broken(e.clone());
697

            
698
                    return Err(e);
699
                }
700
            }
701
        }
702
    }
703

            
704
    /// Run one iteration of the reactor loop.
705
50
    async fn run_once(&mut self) -> Result<ShutdownStatus, FatalError> {
706
50
        let mut netdir_events = self.dir_provider.events();
707

            
708
        // Note: TrackingNow tracks the values it is compared with.
709
        // This is equivalent to sleeping for (until - now) units of time,
710
50
        let upload_rate_lim: TrackingNow = TrackingNow::now(&self.imm.runtime);
711
50
        if let PublishStatus::RateLimited(until) = self.status() {
712
            if upload_rate_lim > until {
713
                // We are no longer rate-limited
714
                self.expire_rate_limit().await?;
715
            }
716
50
        }
717

            
718
50
        let reupload_tracking = TrackingNow::now(&self.imm.runtime);
719
50
        let mut reupload_periods = vec![];
720
        {
721
50
            let mut inner = self.inner.lock().expect("poisoned lock");
722
50
            let inner = &mut *inner;
723
55
            while let Some(reupload) = inner.reupload_timers.peek().copied() {
724
                // First, extract all the timeouts that already elapsed.
725
12
                if reupload.when <= reupload_tracking {
726
5
                    inner.reupload_timers.pop();
727
5
                    reupload_periods.push(reupload.period);
728
5
                } else {
729
                    // We are not ready to schedule any more reuploads.
730
                    //
731
                    // How much we need to sleep is implicitly
732
                    // tracked in reupload_tracking (through
733
                    // the TrackingNow implementation)
734
7
                    break;
735
                }
736
            }
737
        }
738

            
739
        // Check if it's time to schedule any reuploads.
740
50
        for period in reupload_periods {
741
5
            if self.mark_dirty(&period) {
742
5
                debug!(
743
                    time_period=?period,
744
                    "descriptor reupload timer elapsed; scheduling reupload",
745
                );
746
5
                self.update_publish_status_unless_rate_lim(PublishStatus::UploadScheduled)
747
5
                    .await?;
748
            }
749
        }
750

            
751
50
        select_biased! {
752
50
            res = self.upload_task_complete_rx.next().fuse() => {
753
7
                let Some(upload_res) = res else {
754
                    return Ok(ShutdownStatus::Terminate);
755
                };
756

            
757
7
                self.handle_upload_results(upload_res);
758
7
                self.upload_result_to_svc_status()?;
759
            },
760
50
            () = upload_rate_lim.wait_for_earliest(&self.imm.runtime).fuse() => {
761
                self.expire_rate_limit().await?;
762
            },
763
50
            () = reupload_tracking.wait_for_earliest(&self.imm.runtime).fuse() => {
764
                // Run another iteration, executing run_once again. This time, we will remove the
765
                // expired reupload from self.reupload_timers, mark the descriptor dirty for all
766
                // relevant HsDirs, and schedule the upload by setting our status to
767
                // UploadScheduled.
768
5
                return Ok(ShutdownStatus::Continue);
769
            },
770
50
            netdir_event = netdir_events.next().fuse() => {
771
                let Some(netdir_event) = netdir_event else {
772
                    debug!("netdir event stream ended");
773
                    return Ok(ShutdownStatus::Terminate);
774
                };
775

            
776
                if !matches!(netdir_event, DirEvent::NewConsensus) {
777
                    return Ok(ShutdownStatus::Continue);
778
                };
779

            
780
                // The consensus changed. Grab a new NetDir.
781
                let netdir = match self.dir_provider.netdir(Timeliness::Timely) {
782
                    Ok(y) => y,
783
                    Err(e) => {
784
                        error_report!(e, "HS service {}: netdir unavailable. Retrying...", self.imm.nickname);
785
                        // Hopefully a netdir will appear in the future.
786
                        // in the meantime, suspend operations.
787
                        //
788
                        // TODO (#1218): there is a bug here: we stop reading on our inputs
789
                        // including eg publish_status_rx, but it is our job to log some of
790
                        // these things.  While we are waiting for a netdir, all those messages
791
                        // are "stuck"; they'll appear later, with misleading timestamps.
792
                        //
793
                        // Probably this should be fixed by moving the logging
794
                        // out of the reactor, where it won't be blocked.
795
                        self.dir_provider.wait_for_netdir(Timeliness::Timely)
796
                            .await?
797
                    }
798
                };
799
                let relevant_periods = netdir.hs_all_time_periods();
800
                self.handle_consensus_change(netdir).await?;
801
                expire_publisher_keys(
802
                    &self.imm.keymgr,
803
                    &self.imm.nickname,
804
                    &relevant_periods,
805
                ).unwrap_or_else(|e| {
806
                    error_report!(e, "failed to remove expired keys");
807
                });
808
            }
809
50
            update = self.ipt_watcher.await_update().fuse() => {
810
4
                if self.handle_ipt_change(update).await? == ShutdownStatus::Terminate {
811
                    return Ok(ShutdownStatus::Terminate);
812
4
                }
813
            },
814
50
            config = self.config_rx.next().fuse() => {
815
4
                let Some(config) = config else {
816
                    return Ok(ShutdownStatus::Terminate);
817
                };
818

            
819
4
                self.handle_svc_config_change(&config).await?;
820
            },
821
50
            res = self.key_dirs_rx.next().fuse() => {
822
4
                let Some(event) = res else {
823
                    return Ok(ShutdownStatus::Terminate);
824
                };
825

            
826
4
                while let Some(_ignore) = self.key_dirs_rx.try_recv() {
827
                    // Discard other events, so that we only reload once.
828
                }
829

            
830
4
                self.handle_key_dirs_change(event).await?;
831
            }
832
50
            should_upload = self.publish_status_rx.next().fuse() => {
833
22
                let Some(should_upload) = should_upload else {
834
                    return Ok(ShutdownStatus::Terminate);
835
                };
836

            
837
                // Our PublishStatus changed -- are we ready to publish?
838
22
                if should_upload == PublishStatus::UploadScheduled {
839
9
                    self.update_publish_status_unless_waiting(PublishStatus::Idle).await?;
840
9
                    self.upload_all().await?;
841
13
                }
842
            }
843
50
            update_tp_pow_seed = self.update_from_pow_manager_rx.next().fuse() => {
844
                debug!("Update PoW seed for TP!");
845
                let Some(time_period) = update_tp_pow_seed else {
846
                    return Ok(ShutdownStatus::Terminate);
847
                };
848
                self.mark_dirty(&time_period);
849
                self.upload_all().await?;
850
            }
851
        }
852

            
853
41
        Ok(ShutdownStatus::Continue)
854
46
    }
855

            
856
    /// Returns the current status of the publisher
857
86
    fn status(&self) -> PublishStatus {
858
86
        *self.publish_status_rx.borrow()
859
86
    }
860

            
861
    /// Handle a batch of upload outcomes,
862
    /// possibly updating the status of the descriptor for the corresponding HSDirs.
863
7
    fn handle_upload_results(&self, results: TimePeriodUploadResult) {
864
7
        let mut inner = self.inner.lock().expect("poisoned lock");
865
7
        let inner = &mut *inner;
866

            
867
        // Check which time period these uploads pertain to.
868
7
        let period = inner
869
7
            .time_periods
870
7
            .iter_mut()
871
7
            .find(|ctx| ctx.params.time_period() == results.time_period);
872

            
873
7
        let Some(period) = period else {
874
            // The uploads were for a time period that is no longer relevant, so we
875
            // can ignore the result.
876
            return;
877
        };
878

            
879
        // We will need to reupload this descriptor at some point, so we pick
880
        // a random time between 60 minutes and 120 minutes in the future.
881
        //
882
        // See https://spec.torproject.org/rend-spec/deriving-keys.html#WHEN-HSDESC
883
7
        let mut rng = self.imm.mockable.thread_rng();
884
        // TODO SPEC: Control republish period using a consensus parameter?
885
7
        let minutes = rng.gen_range_checked(60..=120).expect("low > high?!");
886
7
        let duration = Duration::from_secs(minutes * 60);
887
7
        let reupload_when = self.imm.runtime.now() + duration;
888
7
        let time_period = period.params.time_period();
889

            
890
7
        info!(
891
            time_period=?time_period,
892
            "reuploading descriptor in {}",
893
            humantime::format_duration(duration),
894
        );
895

            
896
7
        inner.reupload_timers.push(ReuploadTimer {
897
7
            period: time_period,
898
7
            when: reupload_when,
899
7
        });
900

            
901
7
        let mut upload_results = vec![];
902
56
        for upload_res in results.hsdir_result {
903
56
            let relay = period
904
56
                .hs_dirs
905
56
                .iter_mut()
906
252
                .find(|(relay_ids, _status)| relay_ids == &upload_res.relay_ids);
907

            
908
56
            let Some((_relay, status)): Option<&mut (RelayIds, _)> = relay else {
909
                // This HSDir went away, so the result doesn't matter.
910
                // Continue processing the rest of the results
911
                continue;
912
            };
913

            
914
56
            if upload_res.upload_res.is_ok() {
915
56
                let update_last_successful = match period.last_successful {
916
2
                    None => true,
917
54
                    Some(counter) => counter <= upload_res.revision_counter,
918
                };
919

            
920
56
                if update_last_successful {
921
56
                    period.last_successful = Some(upload_res.revision_counter);
922
56
                    // TODO (#1098): Is it possible that this won't update the statuses promptly
923
56
                    // enough. For example, it's possible for the reactor to see a Dirty descriptor
924
56
                    // and start an upload task for a descriptor has already been uploaded (or is
925
56
                    // being uploaded) in another task, but whose upload results have not yet been
926
56
                    // processed.
927
56
                    //
928
56
                    // This is probably made worse by the fact that the statuses are updated in
929
56
                    // batches (grouped by time period), rather than one by one as the upload tasks
930
56
                    // complete (updating the status involves locking the inner mutex, and I wanted
931
56
                    // to minimize the locking/unlocking overheads). I'm not sure handling the
932
56
                    // updates in batches was the correct decision here.
933
56
                    *status = DescriptorStatus::Clean;
934
56
                }
935
            }
936

            
937
56
            upload_results.push(upload_res);
938
        }
939

            
940
7
        period.set_upload_results(upload_results);
941
7
    }
942

            
943
    /// Maybe update our list of HsDirs.
944
    async fn handle_consensus_change(&mut self, netdir: Arc<NetDir>) -> Result<(), FatalError> {
945
        trace!("the consensus has changed; recomputing HSDirs");
946

            
947
        let _old: Option<Arc<NetDir>> = self.replace_netdir(netdir);
948

            
949
        self.recompute_hs_dirs()?;
950
        self.update_publish_status_unless_waiting(PublishStatus::UploadScheduled)
951
            .await?;
952

            
953
        // If the time period has changed, some of our upload results may now be irrelevant,
954
        // so we might need to update our status (for example, if our uploads are
955
        // for a no-longer-relevant time period, it means we might be able to update
956
        // out status from "degraded" to "running")
957
        self.upload_result_to_svc_status()?;
958

            
959
        Ok(())
960
    }
961

            
962
    /// Recompute the HsDirs for all relevant time periods.
963
    fn recompute_hs_dirs(&self) -> Result<(), FatalError> {
964
        let mut inner = self.inner.lock().expect("poisoned lock");
965
        let inner = &mut *inner;
966

            
967
        let netdir = Arc::clone(
968
            inner
969
                .netdir
970
                .as_ref()
971
                .ok_or_else(|| internal!("started upload task without a netdir"))?,
972
        );
973

            
974
        // Update our list of relevant time periods.
975
        let new_time_periods = self.compute_time_periods(&netdir, &inner.time_periods)?;
976
        inner.time_periods = new_time_periods;
977

            
978
        Ok(())
979
    }
980

            
981
    /// Compute the [`TimePeriodContext`]s for the time periods from the specified [`NetDir`].
982
    ///
983
    /// The specified `time_periods` are used to preserve the `DescriptorStatus` of the
984
    /// HsDirs where possible.
985
4
    fn compute_time_periods(
986
4
        &self,
987
4
        netdir: &Arc<NetDir>,
988
4
        time_periods: &[TimePeriodContext],
989
4
    ) -> Result<Vec<TimePeriodContext>, FatalError> {
990
4
        netdir
991
4
            .hs_all_time_periods()
992
4
            .iter()
993
4
            .map(|params| {
994
4
                let period = params.time_period();
995
4
                let blind_id_kp =
996
4
                    read_blind_id_keypair(&self.imm.keymgr, &self.imm.nickname, period)?
997
                        // Note: for now, read_blind_id_keypair cannot return Ok(None).
998
                        // It's supposed to return Ok(None) if we're in offline hsid mode,
999
                        // but that might change when we do #1194
4
                        .ok_or_else(|| internal!("offline hsid mode not supported"))?;
4
                let blind_id: HsBlindIdKey = (&blind_id_kp).into();
                // If our previous `TimePeriodContext`s also had an entry for `period`, we need to
                // preserve the `DescriptorStatus` of its HsDirs. This helps prevent unnecessarily
                // publishing the descriptor to the HsDirs that already have it (the ones that are
                // marked with DescriptorStatus::Clean).
                //
                // In other words, we only want to publish to those HsDirs that
                //   * are part of a new time period (which we have never published the descriptor
                //   for), or
                //   * have just been added to the ring of a time period we already knew about
4
                if let Some(ctx) = time_periods
4
                    .iter()
4
                    .find(|ctx| ctx.params.time_period() == period)
                {
                    TimePeriodContext::new(
                        params.clone(),
                        blind_id.into(),
                        netdir,
                        ctx.hs_dirs.iter(),
                        ctx.upload_results.clone(),
                    )
                } else {
                    // Passing an empty iterator here means all HsDirs in this TimePeriodContext
                    // will be marked as dirty, meaning we will need to upload our descriptor to them.
4
                    TimePeriodContext::new(
4
                        params.clone(),
4
                        blind_id.into(),
4
                        netdir,
4
                        iter::empty(),
4
                        vec![],
                    )
                }
4
            })
4
            .collect::<Result<Vec<TimePeriodContext>, FatalError>>()
4
    }
    /// Replace the old netdir with the new, returning the old.
    fn replace_netdir(&self, new_netdir: Arc<NetDir>) -> Option<Arc<NetDir>> {
        self.inner
            .lock()
            .expect("poisoned lock")
            .netdir
            .replace(new_netdir)
    }
    /// Replace our view of the service config with `new_config` if `new_config` contains changes
    /// that would cause us to generate a new descriptor.
4
    fn replace_config_if_changed(&self, new_config: Arc<OnionServiceConfigPublisherView>) -> bool {
4
        let mut inner = self.inner.lock().expect("poisoned lock");
4
        let old_config = &mut inner.config;
        // The fields we're interested in haven't changed, so there's no need to update
        // `inner.config`.
4
        if *old_config == new_config {
4
            return false;
        }
        let log_change = match (
            old_config.restricted_discovery.enabled,
            new_config.restricted_discovery.enabled,
        ) {
            (true, false) => Some("Disabling restricted discovery mode"),
            (false, true) => Some("Enabling restricted discovery mode"),
            _ => None,
        };
        if let Some(msg) = log_change {
            info!(nickname=%self.imm.nickname, "{}", msg);
        }
        let _old: Arc<OnionServiceConfigPublisherView> = std::mem::replace(old_config, new_config);
        true
4
    }
    /// Recreate the FileWatcher for watching the restricted discovery key_dirs.
8
    fn update_file_watcher(&self) {
8
        let mut inner = self.inner.lock().expect("poisoned lock");
8
        if inner.config.restricted_discovery.watch_configuration() {
            debug!("The restricted_discovery.key_dirs have changed, updating file watcher");
            let mut watcher = FileWatcher::builder(self.imm.runtime.clone());
            let dirs = inner.config.restricted_discovery.key_dirs().clone();
            watch_dirs(&mut watcher, &dirs, &self.path_resolver);
            let watcher = watcher
                .start_watching(self.key_dirs_tx.clone())
                .map_err(|e| {
                    // TODO: update the publish status (see also the module-level TODO about this).
                    error_report!(e, "Cannot set file watcher");
                })
                .ok();
            inner.file_watcher = watcher;
        } else {
8
            if inner.file_watcher.is_some() {
                debug!("removing key_dirs watcher");
8
            }
8
            inner.file_watcher = None;
        }
8
    }
    /// Read the intro points from `ipt_watcher`, and decide whether we're ready to start
    /// uploading.
4
    fn note_ipt_change(&self) -> PublishStatus {
4
        let mut ipts = self.ipt_watcher.borrow_for_publish();
4
        match ipts.ipts.as_mut() {
4
            Some(_ipts) => PublishStatus::UploadScheduled,
            None => PublishStatus::AwaitingIpts,
        }
4
    }
    /// Update our list of introduction points.
4
    async fn handle_ipt_change(
4
        &mut self,
4
        update: Option<Result<(), crate::FatalError>>,
4
    ) -> Result<ShutdownStatus, FatalError> {
4
        trace!(nickname=%self.imm.nickname, "received IPT change notification from IPT manager");
4
        match update {
            Some(Ok(())) => {
4
                let should_upload = self.note_ipt_change();
4
                debug!(nickname=%self.imm.nickname, "the introduction points have changed");
4
                self.mark_all_dirty();
4
                self.update_publish_status_unless_rate_lim(should_upload)
4
                    .await?;
4
                Ok(ShutdownStatus::Continue)
            }
            Some(Err(e)) => Err(e),
            None => {
                debug!(nickname=%self.imm.nickname, "received shut down signal from IPT manager");
                Ok(ShutdownStatus::Terminate)
            }
        }
4
    }
    /// Update the `PublishStatus` of the reactor with `new_state`,
    /// unless the current state is `AwaitingIpts`.
9
    async fn update_publish_status_unless_waiting(
9
        &mut self,
9
        new_state: PublishStatus,
9
    ) -> Result<(), FatalError> {
        // Only update the state if we're not waiting for intro points.
9
        if self.status() != PublishStatus::AwaitingIpts {
9
            self.update_publish_status(new_state).await?;
        }
9
        Ok(())
9
    }
    /// Update the `PublishStatus` of the reactor with `new_state`,
    /// unless the current state is `RateLimited`.
9
    async fn update_publish_status_unless_rate_lim(
9
        &mut self,
9
        new_state: PublishStatus,
9
    ) -> Result<(), FatalError> {
        // We can't exit this state until the rate-limit expires.
9
        if !matches!(self.status(), PublishStatus::RateLimited(_)) {
9
            self.update_publish_status(new_state).await?;
        }
9
        Ok(())
9
    }
    /// Unconditionally update the `PublishStatus` of the reactor with `new_state`.
18
    async fn update_publish_status(&mut self, new_state: PublishStatus) -> Result<(), Bug> {
18
        let onion_status = match new_state {
9
            PublishStatus::Idle => None,
            PublishStatus::UploadScheduled
            | PublishStatus::AwaitingIpts
9
            | PublishStatus::RateLimited(_) => Some(State::Bootstrapping),
        };
18
        if let Some(onion_status) = onion_status {
9
            self.imm.status_tx.send(onion_status, None);
9
        }
18
        trace!(
            "publisher reactor status change: {:?} -> {:?}",
            self.status(),
            new_state
        );
18
        self.publish_status_tx.send(new_state).await.map_err(
            |_: postage::sink::SendError<_>| internal!("failed to send upload notification?!"),
        )?;
18
        Ok(())
18
    }
    /// Update the onion svc status based on the results of the last descriptor uploads.
7
    fn upload_result_to_svc_status(&self) -> Result<(), FatalError> {
7
        let inner = self.inner.lock().expect("poisoned lock");
7
        let netdir = inner
7
            .netdir
7
            .as_ref()
7
            .ok_or_else(|| internal!("handling upload results without netdir?!"))?;
7
        let (state, err) = upload_result_state(netdir, &inner.time_periods);
7
        self.imm.status_tx.send(state, err);
7
        Ok(())
7
    }
    /// Update the descriptors based on the config change.
4
    async fn handle_svc_config_change(
4
        &mut self,
4
        config: &OnionServiceConfig,
4
    ) -> Result<(), FatalError> {
4
        let new_config = Arc::new(config.into());
4
        if self.replace_config_if_changed(Arc::clone(&new_config)) {
            self.update_file_watcher();
            self.update_authorized_clients_if_changed();
            info!(nickname=%self.imm.nickname, "Config has changed, generating a new descriptor");
            self.mark_all_dirty();
            // Schedule an upload, unless we're still waiting for IPTs.
            self.update_publish_status_unless_waiting(PublishStatus::UploadScheduled)
                .await?;
4
        }
4
        Ok(())
4
    }
    /// Update the descriptors based on a restricted discovery key_dirs change.
    ///
    /// If the authorized clients from the [`RestrictedDiscoveryConfig`] have changed,
    /// this marks the descriptor as dirty for all time periods,
    /// and schedules a reupload.
4
    async fn handle_key_dirs_change(&mut self, event: FileEvent) -> Result<(), FatalError> {
4
        debug!("The configured key_dirs have changed");
4
        match event {
4
            FileEvent::Rescan | FileEvent::FileChanged => {
4
                // These events are handled in the same way, by re-reading the keys from disk
4
                // and republishing the descriptor if necessary
4
            }
            _ => return Err(internal!("file watcher event {event:?}").into()),
        };
        // Update the file watcher, in case the change was triggered by a key_dir move.
4
        self.update_file_watcher();
4
        if self.update_authorized_clients_if_changed() {
            self.mark_all_dirty();
            // Schedule an upload, unless we're still waiting for IPTs.
            self.update_publish_status_unless_waiting(PublishStatus::UploadScheduled)
                .await?;
4
        }
4
        Ok(())
4
    }
    /// Recreate the authorized_clients based on the current config.
    ///
    /// Returns `true` if the authorized clients have changed.
4
    fn update_authorized_clients_if_changed(&mut self) -> bool {
4
        let mut inner = self.inner.lock().expect("poisoned lock");
4
        let authorized_clients =
4
            Self::read_authorized_clients(&inner.config.restricted_discovery, &self.path_resolver);
4
        let clients = &mut inner.authorized_clients;
4
        let changed = clients.as_ref() != authorized_clients.as_ref();
4
        if changed {
            info!("The restricted discovery mode authorized clients have changed");
            *clients = authorized_clients;
4
        }
4
        changed
4
    }
    /// Read the authorized `RestrictedDiscoveryKeys` from `config`.
8
    fn read_authorized_clients(
8
        config: &RestrictedDiscoveryConfig,
8
        path_resolver: &CfgPathResolver,
8
    ) -> Option<Arc<RestrictedDiscoveryKeys>> {
8
        let authorized_clients = config.read_keys(path_resolver);
8
        if matches!(authorized_clients.as_ref(), Some(c) if c.is_empty()) {
            warn!(
                "Running in restricted discovery mode, but we have no authorized clients. Service will be unreachable"
            );
8
        }
8
        authorized_clients.map(Arc::new)
8
    }
    /// Mark the descriptor dirty for all time periods.
4
    fn mark_all_dirty(&self) {
4
        trace!("marking the descriptor dirty for all time periods");
4
        self.inner
4
            .lock()
4
            .expect("poisoned lock")
4
            .time_periods
4
            .iter_mut()
4
            .for_each(|tp| tp.mark_all_dirty());
4
    }
    /// Mark the descriptor dirty for the specified time period.
    ///
    /// Returns `true` if the specified period is still relevant, and `false` otherwise.
5
    fn mark_dirty(&self, period: &TimePeriod) -> bool {
5
        let mut inner = self.inner.lock().expect("poisoned lock");
5
        let period_ctx = inner
5
            .time_periods
5
            .iter_mut()
5
            .find(|tp| tp.params.time_period() == *period);
5
        match period_ctx {
5
            Some(ctx) => {
5
                trace!(time_period=?period, "marking the descriptor dirty");
5
                ctx.mark_all_dirty();
5
                true
            }
            None => false,
        }
5
    }
    /// Try to upload our descriptor to the HsDirs that need it.
    ///
    /// If we've recently uploaded some descriptors, we return immediately and schedule the upload
    /// to happen after [`UPLOAD_RATE_LIM_THRESHOLD`].
    ///
    /// Failed uploads are retried
    /// (see [`upload_descriptor_with_retries`](Reactor::upload_descriptor_with_retries)).
    ///
    /// If restricted discovery mode is enabled and there are no authorized clients,
    /// we abort the upload and set our status to [`State::Broken`].
    //
    // Note: a broken restricted discovery config won't prevent future uploads from being scheduled
    // (for example if the IPTs change),
    // which can can cause the publisher's status to oscillate between `Bootstrapping` and `Broken`.
    // TODO: we might wish to refactor the publisher to be more sophisticated about this.
    //
    /// For each current time period, we spawn a task that uploads the descriptor to
    /// all the HsDirs on the HsDir ring of that time period.
    /// Each task shuts down on completion, or when the reactor is dropped.
    ///
    /// Each task reports its upload results (`TimePeriodUploadResult`)
    /// via the `upload_task_complete_tx` channel.
    /// The results are received and processed in the main loop of the reactor.
    ///
    /// Returns an error if it fails to spawn a task, or if an internal error occurs.
9
    async fn upload_all(&mut self) -> Result<(), FatalError> {
9
        trace!("starting descriptor upload task...");
        // Abort the upload entirely if we have an empty list of authorized clients
9
        let authorized_clients = match self.authorized_clients() {
9
            Ok(authorized_clients) => authorized_clients,
            Err(e) => {
                error_report!(e, "aborting upload");
                self.imm.status_tx.send_broken(e.clone());
                // Returning an error would shut down the reactor, so we have to return Ok here.
                return Ok(());
            }
        };
9
        let last_uploaded = self.inner.lock().expect("poisoned lock").last_uploaded;
9
        let now = self.imm.runtime.now();
        // Check if we should rate-limit this upload.
9
        if let Some(ts) = last_uploaded {
5
            let duration_since_upload = now.duration_since(ts);
5
            if duration_since_upload < UPLOAD_RATE_LIM_THRESHOLD {
                return Ok(self.start_rate_limit(UPLOAD_RATE_LIM_THRESHOLD).await?);
5
            }
4
        }
9
        let mut inner = self.inner.lock().expect("poisoned lock");
9
        let inner = &mut *inner;
9
        let _ = inner.last_uploaded.insert(now);
9
        for period_ctx in inner.time_periods.iter_mut() {
9
            let upload_task_complete_tx = self.upload_task_complete_tx.clone();
            // Figure out which HsDirs we need to upload the descriptor to (some of them might already
            // have our latest descriptor, so we filter them out).
9
            let hs_dirs = period_ctx
9
                .hs_dirs
9
                .iter()
72
                .filter_map(|(relay_id, status)| {
72
                    if *status == DescriptorStatus::Dirty {
72
                        Some(relay_id.clone())
                    } else {
                        None
                    }
72
                })
9
                .collect::<Vec<_>>();
9
            if hs_dirs.is_empty() {
                trace!("the descriptor is clean for all HSDirs. Nothing to do");
                return Ok(());
9
            }
9
            let time_period = period_ctx.params.time_period();
            // This scope exists because rng is not Send, so it needs to fall out of scope before we
            // await anything.
9
            let netdir = Arc::clone(
9
                inner
9
                    .netdir
9
                    .as_ref()
9
                    .ok_or_else(|| internal!("started upload task without a netdir"))?,
            );
9
            let imm = Arc::clone(&self.imm);
9
            let ipt_upload_view = self.ipt_watcher.upload_view();
9
            let config = Arc::clone(&inner.config);
9
            let authorized_clients = authorized_clients.clone();
9
            trace!(nickname=%self.imm.nickname, time_period=?time_period,
                "spawning upload task"
            );
9
            let params = period_ctx.params.clone();
9
            let shutdown_rx = self.shutdown_tx.subscribe();
            // Spawn a task to upload the descriptor to all HsDirs of this time period.
            //
            // This task will shut down when the reactor is dropped (i.e. when shutdown_rx is
            // dropped).
9
            let _handle: () = self
9
                .imm
9
                .runtime
9
                .spawn(async move {
9
                    if let Err(e) = Self::upload_for_time_period(
9
                        hs_dirs,
9
                        &netdir,
9
                        config,
9
                        params,
9
                        Arc::clone(&imm),
9
                        ipt_upload_view.clone(),
9
                        authorized_clients.clone(),
9
                        upload_task_complete_tx,
9
                        shutdown_rx,
                    )
9
                    .await
                    {
                        error_report!(
                            e,
                            "descriptor upload failed for HS service {} and time period {:?}",
                            imm.nickname,
                            time_period
                        );
7
                    }
7
                })
9
                .map_err(|e| FatalError::from_spawn("upload_for_time_period task", e))?;
        }
9
        Ok(())
9
    }
    /// Upload the descriptor for the time period specified in `params`.
    ///
    /// Failed uploads are retried
    /// (see [`upload_descriptor_with_retries`](Reactor::upload_descriptor_with_retries)).
    #[allow(clippy::too_many_arguments)] // TODO: refactor
9
    async fn upload_for_time_period(
9
        hs_dirs: Vec<RelayIds>,
9
        netdir: &Arc<NetDir>,
9
        config: Arc<OnionServiceConfigPublisherView>,
9
        params: HsDirParams,
9
        imm: Arc<Immutable<R, M>>,
9
        ipt_upload_view: IptsPublisherUploadView,
9
        authorized_clients: Option<Arc<RestrictedDiscoveryKeys>>,
9
        mut upload_task_complete_tx: mpsc::Sender<TimePeriodUploadResult>,
9
        shutdown_rx: broadcast::Receiver<Void>,
9
    ) -> Result<(), FatalError> {
9
        let time_period = params.time_period();
9
        trace!(time_period=?time_period, "uploading descriptor to all HSDirs for this time period");
9
        let hsdir_count = hs_dirs.len();
        /// An error returned from an upload future.
        //
        // Exhaustive, because this is a private type.
        #[derive(Clone, Debug, thiserror::Error)]
        enum PublishError {
            /// The upload was aborted because there are no IPTs.
            ///
            /// This happens because of an inevitable TOCTOU race, where after being notified by
            /// the IPT manager that the IPTs have changed (via `self.ipt_watcher.await_update`),
            /// we find out there actually are no IPTs, so we can't build the descriptor.
            ///
            /// This is a special kind of error that interrupts the current upload task, and is
            /// logged at `debug!` level rather than `warn!` or `error!`.
            ///
            /// Ideally, this shouldn't happen very often (if at all).
            #[error("No IPTs")]
            NoIpts,
            /// The reactor has shut down
            #[error("The reactor has shut down")]
            Shutdown,
            /// An fatal error.
            #[error("{0}")]
            Fatal(#[from] FatalError),
        }
9
        let max_hsdesc_len: usize = netdir
9
            .params()
9
            .hsdir_max_desc_size
9
            .try_into()
9
            .expect("Unable to convert positive int32 to usize!?");
9
        let upload_results = futures::stream::iter(hs_dirs)
72
            .map(|relay_ids| {
72
                let netdir = netdir.clone();
72
                let config = Arc::clone(&config);
72
                let imm = Arc::clone(&imm);
72
                let ipt_upload_view = ipt_upload_view.clone();
72
                let authorized_clients = authorized_clients.clone();
72
                let params = params.clone();
72
                let mut shutdown_rx = shutdown_rx.clone();
72
                let ed_id = relay_ids
72
                    .rsa_identity()
72
                    .map(|id| id.to_string())
72
                    .unwrap_or_else(|| "unknown".into());
72
                let rsa_id = relay_ids
72
                    .rsa_identity()
72
                    .map(|id| id.to_string())
72
                    .unwrap_or_else(|| "unknown".into());
72
                async move {
72
                    let run_upload = |desc| async {
72
                        let Some(hsdir) = netdir.by_ids(&relay_ids) else {
                            // This should never happen (all of our relay_ids are from the stored
                            // netdir).
                            let err =
                                "tried to upload descriptor to relay not found in consensus?!";
                            warn!(
                                nickname=%imm.nickname, hsdir_id=%ed_id, hsdir_rsa_id=%rsa_id,
                                "{err}"
                            );
                            return Err(internal!("{err}").into());
                        };
72
                        Self::upload_descriptor_with_retries(
72
                            desc,
72
                            &netdir,
72
                            &hsdir,
72
                            &ed_id,
72
                            &rsa_id,
72
                            Arc::clone(&imm),
72
                        )
72
                        .await
128
                    };
                    // How long until we're supposed to time out?
72
                    let worst_case_end = imm.runtime.now() + OVERALL_UPLOAD_TIMEOUT;
                    // We generate a new descriptor before _each_ HsDir upload. This means each
                    // HsDir could, in theory, receive a different descriptor (not just in terms of
                    // revision-counters, but also with a different set of IPTs). It may seem like
                    // this could lead to some HsDirs being left with an outdated descriptor, but
                    // that's not the case: after the upload completes, the publisher will be
                    // notified by the ipt_watcher of the IPT change event (if there was one to
                    // begin with), which will trigger another upload job.
72
                    let hsdesc = {
                        // This scope is needed because the ipt_set MutexGuard is not Send, so it
                        // needs to fall out of scope before the await point below
72
                        let mut ipt_set = ipt_upload_view.borrow_for_publish();
                        // If there are no IPTs, we abort the upload. At this point, we might have
                        // uploaded the descriptor to some, but not all, HSDirs from the specified
                        // time period.
                        //
                        // Returning an error here means the upload completion task is never
                        // notified of the outcome of any of these uploads (which means the
                        // descriptor is not marked clean). This is OK, because if we suddenly find
                        // out we have no IPTs, it means our built `hsdesc` has an outdated set of
                        // IPTs, so we need to go back to the main loop to wait for IPT changes,
                        // and generate a fresh descriptor anyway.
                        //
                        // Ideally, this shouldn't happen very often (if at all).
72
                        let Some(ipts) = ipt_set.ipts.as_mut() else {
                            return Err(PublishError::NoIpts);
                        };
72
                        let hsdesc = {
72
                            trace!(
                                nickname=%imm.nickname, time_period=?time_period,
                                "building descriptor"
                            );
72
                            let mut rng = imm.mockable.thread_rng();
72
                            let mut key_rng = tor_llcrypto::rng::CautiousRng;
                            // We're about to generate a new version of the descriptor,
                            // so let's generate a new revision counter.
72
                            let now = imm.runtime.wallclock();
72
                            let revision_counter = imm.generate_revision_counter(&params, now)?;
72
                            build_sign(
72
                                &imm.keymgr,
72
                                &imm.pow_manager,
72
                                &config,
72
                                netdir.params(),
72
                                authorized_clients.as_deref(),
72
                                ipts,
72
                                time_period,
72
                                revision_counter,
72
                                &mut rng,
72
                                &mut key_rng,
72
                                imm.runtime.wallclock(),
72
                                max_hsdesc_len,
                            )?
                        };
                        if let Err(e) =
72
                            ipt_set.note_publication_attempt(&imm.runtime, worst_case_end)
                        {
                            let wait = e.log_retry_max(&imm.nickname)?;
                            // TODO (#1226): retry instead of this
                            return Err(FatalError::Bug(internal!(
                                "ought to retry after {wait:?}, crashing instead"
                            ))
                            .into());
72
                        }
72
                        hsdesc
                    };
                    let VersionedDescriptor {
72
                        desc,
72
                        revision_counter,
72
                    } = hsdesc;
72
                    let desc: Arc<str> = desc.into();
72
                    trace!(
                        nickname=%imm.nickname, time_period=?time_period,
                        revision_counter=?revision_counter,
                        "generated new descriptor for time period",
                    );
                    // (Actually launch the upload attempt. No timeout is needed
                    // here, since the backoff::Runner code will handle that for us.)
72
                    let upload_res: UploadResult = select_biased! {
72
                        shutdown = shutdown_rx.next().fuse() => {
                            // This will always be None, since Void is uninhabited.
                            let _: Option<Void> = shutdown;
                            // It looks like the reactor has shut down,
                            // so there is no point in uploading the descriptor anymore.
                            //
                            // Let's shut down the upload task too.
                            trace!(
                                nickname=%imm.nickname, time_period=?time_period,
                                "upload task received shutdown signal"
                            );
                            return Err(PublishError::Shutdown);
                        },
72
                        res = run_upload(desc.clone()).fuse() => res,
                    };
                    // Note: UploadResult::Failure is only returned when
                    // upload_descriptor_with_retries fails, i.e. if all our retry
                    // attempts have failed
56
                    Ok(HsDirUploadStatus {
56
                        relay_ids,
56
                        upload_res,
56
                        revision_counter,
56
                    })
56
                }
72
            })
            // This fails to compile unless the stream is boxed. See https://github.com/rust-lang/rust/issues/104382
9
            .boxed()
9
            .buffer_unordered(MAX_CONCURRENT_UPLOADS)
9
            .try_collect::<Vec<_>>()
9
            .await;
7
        let upload_results = match upload_results {
7
            Ok(v) => v,
            Err(PublishError::Fatal(e)) => return Err(e),
            Err(PublishError::NoIpts) => {
                debug!(
                    nickname=%imm.nickname, time_period=?time_period,
                     "no introduction points; skipping upload"
                );
                return Ok(());
            }
            Err(PublishError::Shutdown) => {
                debug!(
                    nickname=%imm.nickname, time_period=?time_period,
                     "the reactor has shut down; aborting upload"
                );
                return Ok(());
            }
        };
7
        let (succeeded, _failed): (Vec<_>, Vec<_>) = upload_results
7
            .iter()
56
            .partition(|res| res.upload_res.is_ok());
7
        debug!(
            nickname=%imm.nickname, time_period=?time_period,
            "descriptor uploaded successfully to {}/{} HSDirs",
            succeeded.len(), hsdir_count
        );
7
        if upload_task_complete_tx
7
            .send(TimePeriodUploadResult {
7
                time_period,
7
                hsdir_result: upload_results,
7
            })
7
            .await
7
            .is_err()
        {
            return Err(internal!(
                "failed to notify reactor of upload completion (reactor shut down)"
            )
            .into());
7
        }
7
        Ok(())
7
    }
    /// Upload a descriptor to the specified HSDir.
    ///
    /// If an upload fails, this returns an `Err`. This function does not handle retries. It is up
    /// to the caller to retry on failure.
    ///
    /// This function does not handle timeouts.
88
    async fn upload_descriptor(
88
        hsdesc: Arc<str>,
88
        netdir: &Arc<NetDir>,
88
        hsdir: &Relay<'_>,
88
        imm: Arc<Immutable<R, M>>,
88
    ) -> Result<(), UploadError> {
88
        let request = HsDescUploadRequest::new(hsdesc);
88
        trace!(nickname=%imm.nickname, hsdir_id=%hsdir.id(), hsdir_rsa_id=%hsdir.rsa_id(),
            "starting descriptor upload",
        );
88
        let tunnel = imm
88
            .mockable
88
            .get_or_launch_hs_dir(netdir, OwnedCircTarget::from_circ_target(hsdir))
88
            .await?;
88
        let source: Option<SourceInfo> = tunnel
88
            .source_info()
88
            .map_err(into_internal!("Couldn't get SourceInfo for circuit"))?;
88
        let mut stream = tunnel
88
            .begin_dir_stream()
88
            .await
88
            .map_err(UploadError::Stream)?;
88
        let _response: String = send_request(&imm.runtime, &request, &mut stream, source)
88
            .await
88
            .map_err(|dir_error| -> UploadError {
16
                match dir_error {
16
                    DirClientError::RequestFailed(e) => e.into(),
                    DirClientError::CircMgr(e) => into_internal!(
                        "tor-dirclient complains about circmgr going wrong but we gave it a stream"
                    )(e)
                    .into(),
                    e => into_internal!("unexpected error")(e).into(),
                }
16
            })?
72
            .into_output_string()?; // This returns an error if we received an error response
56
        Ok(())
88
    }
    /// Upload a descriptor to the specified HSDir, retrying if appropriate.
    ///
    /// Any failed uploads are retried according to a [`PublisherBackoffSchedule`].
    /// Each failed upload is retried until it succeeds, or until the overall timeout specified
    /// by [`BackoffSchedule::overall_timeout`] elapses. Individual attempts are timed out
    /// according to the [`BackoffSchedule::single_attempt_timeout`].
    /// This function gives up after the overall timeout elapses,
    /// declaring the upload a failure, and never retrying it again.
    ///
    /// See also [`BackoffSchedule`].
72
    async fn upload_descriptor_with_retries(
72
        hsdesc: Arc<str>,
72
        netdir: &Arc<NetDir>,
72
        hsdir: &Relay<'_>,
72
        ed_id: &str,
72
        rsa_id: &str,
72
        imm: Arc<Immutable<R, M>>,
72
    ) -> UploadResult {
        /// The base delay to use for the backoff schedule.
        const BASE_DELAY_MSEC: u32 = 1000;
72
        let schedule = PublisherBackoffSchedule {
72
            retry_delay: RetryDelay::from_msec(BASE_DELAY_MSEC),
72
            mockable: imm.mockable.clone(),
72
        };
72
        let runner = Runner::new(
72
            "upload a hidden service descriptor".into(),
72
            schedule.clone(),
72
            imm.runtime.clone(),
        );
88
        let fallible_op = || async {
88
            let r = Self::upload_descriptor(hsdesc.clone(), netdir, hsdir, Arc::clone(&imm)).await;
88
            if let Err(e) = &r {
32
                if e.should_report_as_suspicious() {
                    // Note that not every protocol violation is suspicious:
                    // we only warn on the protocol violations that look like attempts
                    // to do a traffic tagging attack via hsdir inflation.
                    // (See proposal 360.)
                    warn_report!(
                        e,
                        "Suspicious error while uploading descriptor to {}/{}",
                        ed_id,
                        rsa_id
                    );
32
                }
32
                if let UploadError::Request(req_err) = e {
32
                    if let RequestError::HttpStatus(status, _msg) = &req_err.error {
16
                        if (400..500).contains(status) {
                            // This should never happen (it's a bug if it does!)
                            log_ratelim!(
                                "The peer says our HS descriptor upload request was invalid. This is a bug (maybe theirs; maybe ours)";
                                Err::<(), _>(e.clone());
                            );
16
                        }
                        // C Tor HsDirs return a 400 status code if they deem
                        // the descriptor to be invalid.
16
                        if *status == 400 {
                            let res = dump_invalid_hsdesc(&hsdesc, &imm);
                            log_ratelim!(
                                "Dumping rejected HS descriptor to {}",
                                imm.bad_hsdescs_dir.as_path().display_lossy();
                                res;
                            );
                            // TODO: we might want to garbage-collect this at some point
                            // (e.g. if the mtime of the last dumped
                            // bad descriptor is far enough into the past?)
16
                        }
16
                    }
                }
56
            }
88
            r
176
        };
72
        let outcome: Result<(), BackoffError<UploadError>> = runner.run(fallible_op).await;
56
        match outcome {
            Ok(()) => {
56
                debug!(
                    nickname=%imm.nickname, hsdir_id=%ed_id, hsdir_rsa_id=%rsa_id,
                    "successfully uploaded descriptor to HSDir",
                );
56
                Ok(())
            }
            Err(e) => {
                warn_report!(
                    e,
                    "failed to upload descriptor for service {} (hsdir_id={}, hsdir_rsa_id={})",
                    imm.nickname,
                    ed_id,
                    rsa_id
                );
                Err(e.into())
            }
        }
56
    }
    /// Stop publishing descriptors until the specified delay elapses.
    async fn start_rate_limit(&mut self, delay: Duration) -> Result<(), Bug> {
        if !matches!(self.status(), PublishStatus::RateLimited(_)) {
            debug!(
                "We are rate-limited for {}; pausing descriptor publication",
                humantime::format_duration(delay)
            );
            let until = self.imm.runtime.now() + delay;
            self.update_publish_status(PublishStatus::RateLimited(until))
                .await?;
        }
        Ok(())
    }
    /// Handle the upload rate-limit being lifted.
    async fn expire_rate_limit(&mut self) -> Result<(), Bug> {
        debug!("We are no longer rate-limited; resuming descriptor publication");
        self.update_publish_status(PublishStatus::UploadScheduled)
            .await?;
        Ok(())
    }
    /// Return the authorized clients, if restricted mode is enabled.
    ///
    /// Returns `Ok(None)` if restricted discovery mode is disabled.
    ///
    /// Returns an error if restricted discovery mode is enabled, but the client list is empty.
    #[cfg_attr(
        not(feature = "restricted-discovery"),
        allow(clippy::unnecessary_wraps)
    )]
9
    fn authorized_clients(&self) -> Result<Option<Arc<RestrictedDiscoveryKeys>>, FatalError> {
        cfg_if::cfg_if! {
            if #[cfg(feature = "restricted-discovery")] {
9
                let authorized_clients = self
9
                    .inner
9
                    .lock()
9
                    .expect("poisoned lock")
9
                    .authorized_clients
9
                    .clone();
9
                if authorized_clients.as_ref().as_ref().map(|v| v.is_empty()).unwrap_or_default() {
                    return Err(FatalError::RestrictedDiscoveryNoClients);
9
                }
9
                Ok(authorized_clients)
            } else {
                Ok(None)
            }
        }
9
    }
}
/// Dump an invalid HS descriptor to disk, for debugging purposes.
///
/// The descriptor will get written to the state directory,
/// in `hss/<HS_NICKNAME>/bad_hsdescs/current`.
///
// TODO(#2760): this is a quick and dirty hack for debugging #2760.
// We may want to eventually replace this with a more principled
// "invalid descriptor" dumping utility, like the one C Tor has
// in dirparse/unparseable.c.
fn dump_invalid_hsdesc<R, M>(
    hsdesc: &Arc<str>,
    imm: &Arc<Immutable<R, M>>,
) -> Result<(), HsDescDumpError>
where
    R: Runtime,
    M: Mockable,
{
    use rand::distr::{Alphanumeric, SampleString};
    // If multiple upload tasks have an invalid
    // descriptor, they will all race to write to "filename.tmp",
    // so we need a unique id for the temporary file
    // (TODO: maybe it would be useful to have a FileAccess API for this?):
    //
    // Case-insensitive file systems are going to have a higher risk
    // of tmpname collisions for the tasks trying to write an invalid descriptor,
    // but we can try mitigate that by generating a relatively long string
    // (in practice, the chances of this happening are going to be very slim).
    let mut rng = imm.mockable.thread_rng();
    let tmpname = Alphanumeric.sample_string(&mut rng, 16);
    imm.bad_hsdescs_dir
        .write_and_replace(&tmpname, hsdesc.as_ref())?;
    let tmp_path = imm.bad_hsdescs_dir.as_path().join(tmpname);
    let target = imm.bad_hsdescs_dir.as_path().join("current");
    // This rename() bypasses the fs-mistrust checks, but I believe that's okay,
    // because the write_and_replace() above already checked the permissions
    // of the parent dir.
    let res = std::fs::rename(&tmp_path, &target).map_err(|err| HsDescDumpError::Rename {
        filename: target,
        err: Arc::new(err),
    });
    if res.is_err() {
        // Try to at least clean up the temp file...
        let _ = std::fs::remove_file(tmp_path);
    }
    Ok(())
}
/// An error returned when trying to dump an invalid HS desc to disk.
#[derive(Clone, Debug, thiserror::Error)]
enum HsDescDumpError {
    /// We encountered a problem while inspecting or creating a directory.
    #[error("Problem accessing filesystem")]
    FsMistrust(#[from] fs_mistrust::Error),
    /// We encountered an error while attempting an IO operation on a file.
    #[error(r#"IO error while attempting to rename file to {}"#, filename.anonymize_home())]
    Rename {
        /// The file that we were trying to modify or inspect
        filename: PathBuf,
        /// The error that we got when trying to perform the operation.
        #[source]
        err: Arc<std::io::Error>,
    },
}
/// Try to expand a path, logging a warning on failure.
fn maybe_expand_path(p: &CfgPath, r: &CfgPathResolver) -> Option<PathBuf> {
    // map_err returns unit for clarity
    #[allow(clippy::unused_unit, clippy::semicolon_if_nothing_returned)]
    p.path(r)
        .map_err(|e| {
            tor_error::warn_report!(e, "invalid path");
            ()
        })
        .ok()
}
/// Add `path` to the specified `watcher`.
macro_rules! watch_path {
    ($watcher:expr, $path:expr, $watch_fn:ident, $($watch_fn_args:expr,)*) => {{
        if let Err(e) = $watcher.$watch_fn(&$path, $($watch_fn_args)*) {
            warn_report!(e, "failed to watch path {:?}", $path);
        } else {
            debug!("watching path {:?}", $path);
        }
    }}
}
/// Add the specified directories to the watcher.
fn watch_dirs<R: Runtime>(
    watcher: &mut FileWatcherBuilder<R>,
    dirs: &DirectoryKeyProviderList,
    path_resolver: &CfgPathResolver,
) {
    for path in dirs {
        let path = path.path();
        let Some(path) = maybe_expand_path(path, path_resolver) else {
            warn!("failed to expand key_dir path {:?}", path);
            continue;
        };
        // If the path doesn't exist, the notify watcher will return an error if we attempt to watch it,
        // so we skip over paths that don't exist at this time
        // (this obviously suffers from a TOCTOU race, but most of the time,
        // it is good enough at preventing the watcher from failing to watch.
        // If the race *does* happen it is not disastrous, i.e. the reactor won't crash,
        // but it will fail to set the watcher).
        if matches!(path.try_exists(), Ok(true)) {
            watch_path!(watcher, &path, watch_dir, "auth",);
        }
        // FileWatcher::watch_path causes the parent dir of the path to be watched.
        if matches!(path.parent().map(|p| p.try_exists()), Some(Ok(true))) {
            watch_path!(watcher, &path, watch_path,);
        }
    }
}
/// Try to read the blinded identity key for a given `TimePeriod`.
///
/// Returns `None` if the service is running in "offline" mode.
///
// TODO (#1194): we don't currently have support for "offline" mode so this can never return
// `Ok(None)`.
148
pub(super) fn read_blind_id_keypair(
148
    keymgr: &Arc<KeyMgr>,
148
    nickname: &HsNickname,
148
    period: TimePeriod,
148
) -> Result<Option<HsBlindIdKeypair>, FatalError> {
148
    let svc_key_spec = HsIdKeypairSpecifier::new(nickname.clone());
148
    let hsid_kp = keymgr
148
        .get::<HsIdKeypair>(&svc_key_spec)?
148
        .ok_or_else(|| FatalError::MissingHsIdKeypair(nickname.clone()))?;
148
    let blind_id_key_spec = BlindIdKeypairSpecifier::new(nickname.clone(), period);
    // TODO: make the keystore selector configurable
148
    let keystore_selector = Default::default();
148
    match keymgr.get::<HsBlindIdKeypair>(&blind_id_key_spec)? {
148
        Some(kp) => Ok(Some(kp)),
        None => {
            let (_hs_blind_id_key, hs_blind_id_kp, _subcredential) = hsid_kp
                .compute_blinded_key(period)
                .map_err(|_| internal!("failed to compute blinded key"))?;
            // Note: we can't use KeyMgr::generate because this key is derived from the HsId
            // (KeyMgr::generate uses the tor_keymgr::Keygen trait under the hood,
            // which assumes keys are randomly generated, rather than derived from existing keys).
            keymgr.insert(hs_blind_id_kp, &blind_id_key_spec, keystore_selector, true)?;
            let arti_path = |spec: &dyn KeySpecifier| {
                spec.arti_path()
                    .map_err(into_internal!("invalid key specifier?!"))
            };
            Ok(Some(
                keymgr.get::<HsBlindIdKeypair>(&blind_id_key_spec)?.ok_or(
                    FatalError::KeystoreRace {
                        action: "read",
                        path: arti_path(&blind_id_key_spec)?,
                    },
                )?,
            ))
        }
    }
148
}
/// Determine the [`State`] of the publisher based on the upload results
/// from the current `time_periods`.
15
fn upload_result_state(
15
    netdir: &NetDir,
15
    time_periods: &[TimePeriodContext],
15
) -> (State, Option<Problem>) {
15
    let current_period = netdir.hs_time_period();
15
    let current_period_res = time_periods
15
        .iter()
15
        .find(|ctx| ctx.params.time_period() == current_period);
15
    let succeeded_current_tp = current_period_res
15
        .iter()
15
        .flat_map(|res| &res.upload_results)
118
        .filter(|res| res.upload_res.is_ok())
15
        .collect_vec();
15
    let secondary_tp_res = time_periods
15
        .iter()
21
        .filter(|ctx| ctx.params.time_period() != current_period)
15
        .collect_vec();
15
    let succeeded_secondary_tp = secondary_tp_res
15
        .iter()
15
        .flat_map(|res| &res.upload_results)
50
        .filter(|res| res.upload_res.is_ok())
15
        .collect_vec();
    // All of the failed uploads (for all TPs)
15
    let failed = time_periods
15
        .iter()
15
        .flat_map(|res| &res.upload_results)
168
        .filter(|res| res.upload_res.is_err())
15
        .collect_vec();
15
    let problems: Vec<DescUploadRetryError> = failed
15
        .iter()
50
        .flat_map(|e| e.upload_res.as_ref().map_err(|e| e.clone()).err())
15
        .collect();
15
    let err = match problems.as_slice() {
15
        [_, ..] => Some(problems.into()),
11
        [] => None,
    };
15
    if time_periods.len() < 2 {
        // We need at least TP contexts (one for the primary TP,
        // and another for the secondary one).
        //
        // If either is missing, we are unreachable for some or all clients.
9
        return (State::DegradedUnreachable, err);
6
    }
6
    let state = match (
6
        succeeded_current_tp.as_slice(),
6
        succeeded_secondary_tp.as_slice(),
    ) {
6
        (&[], &[..]) | (&[..], &[]) if failed.is_empty() => {
            // We don't have any upload results for one or both TPs.
            // We are still bootstrapping.
3
            State::Bootstrapping
        }
2
        (&[_, ..], &[_, ..]) if failed.is_empty() => {
            // We have uploaded the descriptor to one or more HsDirs from both
            // HsDir rings (primary and secondary), and none of the uploads failed.
            // We are fully reachable.
1
            State::Running
        }
        (&[_, ..], &[_, ..]) => {
            // We have uploaded the descriptor to one or more HsDirs from both
            // HsDir rings (primary and secondary), but some of the uploads failed.
            // We are reachable, but we failed to upload the descriptor to all the HsDirs
            // that were supposed to have it.
1
            State::DegradedReachable
        }
        (&[..], &[]) | (&[], &[..]) => {
            // We have either
            //   * uploaded the descriptor to some of the HsDirs from one of the rings,
            //   but haven't managed to upload it to any of the HsDirs on the other ring, or
            //   * all of the uploads failed
            //
            // Either way, we are definitely not reachable by all clients.
1
            State::DegradedUnreachable
        }
    };
6
    (state, err)
15
}
/// Whether the reactor should initiate an upload.
#[derive(Copy, Clone, Debug, Default, PartialEq)]
enum PublishStatus {
    /// We need to call upload_all.
    UploadScheduled,
    /// We are rate-limited until the specified [`Instant`].
    ///
    /// We have tried to schedule multiple uploads in a short time span,
    /// and we are rate-limited. We are waiting for a signal from the schedule_upload_tx
    /// channel to unblock us.
    RateLimited(Instant),
    /// We are idle and waiting for external events.
    ///
    /// We have enough information to build the descriptor, but since we have already called
    /// upload_all to upload it to all relevant HSDirs, there is nothing for us to do right nbow.
    Idle,
    /// We are waiting for the IPT manager to establish some introduction points.
    ///
    /// No descriptors will be published until the `PublishStatus` of the reactor is changed to
    /// `UploadScheduled`.
    #[default]
    AwaitingIpts,
}
/// The backoff schedule for the task that publishes descriptors.
#[derive(Clone, Debug)]
struct PublisherBackoffSchedule<M: Mockable> {
    /// The delays
    retry_delay: RetryDelay,
    /// The mockable reactor state, needed for obtaining an rng.
    mockable: M,
}
impl<M: Mockable> BackoffSchedule for PublisherBackoffSchedule<M> {
88
    fn max_retries(&self) -> Option<usize> {
88
        None
88
    }
72
    fn overall_timeout(&self) -> Option<Duration> {
72
        Some(OVERALL_UPLOAD_TIMEOUT)
72
    }
88
    fn single_attempt_timeout(&self) -> Option<Duration> {
88
        Some(self.mockable.estimate_upload_timeout())
88
    }
32
    fn next_delay<E: RetriableError>(&mut self, _error: &E) -> Option<Duration> {
32
        Some(self.retry_delay.next_delay(&mut self.mockable.thread_rng()))
32
    }
}
impl RetriableError for UploadError {
64
    fn should_retry(&self) -> bool {
64
        match self {
64
            UploadError::Request(_) | UploadError::Circuit(_) | UploadError::Stream(_) => true,
            UploadError::Bug(_) => false,
        }
64
    }
}
/// The outcome of uploading a descriptor to the HSDirs from a particular time period.
#[derive(Debug, Clone)]
struct TimePeriodUploadResult {
    /// The time period.
    time_period: TimePeriod,
    /// The upload results.
    hsdir_result: Vec<HsDirUploadStatus>,
}
/// The outcome of uploading a descriptor to a particular HsDir.
#[derive(Clone, Debug)]
struct HsDirUploadStatus {
    /// The identity of the HsDir we attempted to upload the descriptor to.
    relay_ids: RelayIds,
    /// The outcome of this attempt.
    upload_res: UploadResult,
    /// The revision counter of the descriptor we tried to upload.
    revision_counter: RevisionCounter,
}
/// The outcome of uploading a descriptor.
type UploadResult = Result<(), DescUploadRetryError>;
impl From<BackoffError<UploadError>> for DescUploadRetryError {
    fn from(e: BackoffError<UploadError>) -> Self {
        use BackoffError as BE;
        use DescUploadRetryError as DURE;
        match e {
            BE::FatalError(e) => DURE::FatalError(e),
            BE::MaxRetryCountExceeded(e) => DURE::MaxRetryCountExceeded(e),
            BE::Timeout(e) => DURE::Timeout(e),
            BE::ExplicitStop(_) => {
                DURE::Bug(internal!("explicit stop in publisher backoff schedule?!"))
            }
        }
    }
}
// NOTE: the rest of the publisher tests live in publish.rs
#[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 tor_netdir::testnet;
    /// Create a `TimePeriodContext` from the specified upload results.
    fn create_time_period_ctx(
        params: &HsDirParams,
        upload_results: Vec<HsDirUploadStatus>,
    ) -> TimePeriodContext {
        TimePeriodContext {
            params: params.clone(),
            hs_dirs: vec![],
            last_successful: None,
            upload_results,
        }
    }
    /// Create a single `HsDirUploadStatus`
    fn create_upload_status(upload_res: UploadResult) -> HsDirUploadStatus {
        HsDirUploadStatus {
            relay_ids: RelayIds::empty(),
            upload_res,
            revision_counter: RevisionCounter::from(13),
        }
    }
    /// Create a bunch of results, all with the specified `upload_res`.
    fn create_upload_results(upload_res: UploadResult) -> Vec<HsDirUploadStatus> {
        std::iter::repeat_with(|| create_upload_status(upload_res.clone()))
            .take(10)
            .collect()
    }
    fn construct_netdir() -> NetDir {
        const SRV1: [u8; 32] = *b"The door refused to open.       ";
        const SRV2: [u8; 32] = *b"It said, 'Five cents, please.'  ";
        let dir = testnet::construct_custom_netdir(|_, _, bld| {
            bld.shared_rand_cur(7, SRV1.into(), None)
                .shared_rand_prev(7, SRV2.into(), None);
        })
        .unwrap();
        dir.unwrap_if_sufficient().unwrap()
    }
    #[test]
    fn upload_result_status_bootstrapping() {
        let netdir = construct_netdir();
        let all_params = netdir.hs_all_time_periods();
        let current_period = netdir.hs_time_period();
        let primary_params = all_params
            .iter()
            .find(|param| param.time_period() == current_period)
            .unwrap();
        let results = [
            (vec![], vec![]),
            (vec![], create_upload_results(Ok(()))),
            (create_upload_results(Ok(())), vec![]),
        ];
        for (primary_result, secondary_result) in results {
            let primary_ctx = create_time_period_ctx(primary_params, primary_result);
            let secondary_params = all_params
                .iter()
                .find(|param| param.time_period() != current_period)
                .unwrap();
            let secondary_ctx = create_time_period_ctx(secondary_params, secondary_result.clone());
            let (status, err) = upload_result_state(&netdir, &[primary_ctx, secondary_ctx]);
            assert_eq!(status, State::Bootstrapping);
            assert!(err.is_none());
        }
    }
    #[test]
    fn upload_result_status_running() {
        let netdir = construct_netdir();
        let all_params = netdir.hs_all_time_periods();
        let current_period = netdir.hs_time_period();
        let primary_params = all_params
            .iter()
            .find(|param| param.time_period() == current_period)
            .unwrap();
        let secondary_result = create_upload_results(Ok(()));
        let secondary_params = all_params
            .iter()
            .find(|param| param.time_period() != current_period)
            .unwrap();
        let secondary_ctx = create_time_period_ctx(secondary_params, secondary_result.clone());
        let primary_result = create_upload_results(Ok(()));
        let primary_ctx = create_time_period_ctx(primary_params, primary_result);
        let (status, err) = upload_result_state(&netdir, &[primary_ctx, secondary_ctx]);
        assert_eq!(status, State::Running);
        assert!(err.is_none());
    }
    #[test]
    fn upload_result_status_reachable() {
        let netdir = construct_netdir();
        let all_params = netdir.hs_all_time_periods();
        let current_period = netdir.hs_time_period();
        let primary_params = all_params
            .iter()
            .find(|param| param.time_period() == current_period)
            .unwrap();
        let primary_result = create_upload_results(Ok(()));
        let primary_ctx = create_time_period_ctx(primary_params, primary_result.clone());
        let failed_res = create_upload_results(Err(DescUploadRetryError::Bug(internal!("test"))));
        let secondary_result = create_upload_results(Ok(()))
            .into_iter()
            .chain(failed_res.iter().cloned())
            .collect();
        let secondary_params = all_params
            .iter()
            .find(|param| param.time_period() != current_period)
            .unwrap();
        let secondary_ctx = create_time_period_ctx(secondary_params, secondary_result);
        let (status, err) = upload_result_state(&netdir, &[primary_ctx, secondary_ctx]);
        // Degraded but reachable (because some of the secondary HsDir uploads failed).
        assert_eq!(status, State::DegradedReachable);
        assert!(matches!(err, Some(Problem::DescriptorUpload(_))));
    }
    #[test]
    fn upload_result_status_unreachable() {
        let netdir = construct_netdir();
        let all_params = netdir.hs_all_time_periods();
        let current_period = netdir.hs_time_period();
        let primary_params = all_params
            .iter()
            .find(|param| param.time_period() == current_period)
            .unwrap();
        let mut primary_result =
            create_upload_results(Err(DescUploadRetryError::Bug(internal!("test"))));
        let primary_ctx = create_time_period_ctx(primary_params, primary_result.clone());
        // No secondary TP (we are unreachable).
        let (status, err) = upload_result_state(&netdir, &[primary_ctx]);
        assert_eq!(status, State::DegradedUnreachable);
        assert!(matches!(err, Some(Problem::DescriptorUpload(_))));
        // Add a successful result
        primary_result.push(create_upload_status(Ok(())));
        let primary_ctx = create_time_period_ctx(primary_params, primary_result.clone());
        let (status, err) = upload_result_state(&netdir, &[primary_ctx]);
        // Still degraded, and unreachable (because we don't have a TimePeriodContext
        // for the secondary TP)
        assert_eq!(status, State::DegradedUnreachable);
        assert!(matches!(err, Some(Problem::DescriptorUpload(_))));
        // If we add another time period where none of the uploads were successful,
        // we're *still* unreachable
        let secondary_result =
            create_upload_results(Err(DescUploadRetryError::Bug(internal!("test"))));
        let secondary_params = all_params
            .iter()
            .find(|param| param.time_period() != current_period)
            .unwrap();
        let secondary_ctx = create_time_period_ctx(secondary_params, secondary_result.clone());
        let primary_ctx = create_time_period_ctx(primary_params, primary_result.clone());
        let (status, err) = upload_result_state(&netdir, &[primary_ctx, secondary_ctx]);
        assert_eq!(status, State::DegradedUnreachable);
        assert!(matches!(err, Some(Problem::DescriptorUpload(_))));
    }
}