tor_hsservice/
publish.rs

1//! Publish and maintain onion service descriptors
2//!
3//! See the [`reactor`] module-level documentation for more details.
4
5mod backoff;
6mod descriptor;
7mod reactor;
8mod reupload_timer;
9
10use crate::config::restricted_discovery::RestrictedDiscoveryKeys;
11use crate::internal_prelude::*;
12use crate::pow::PowManager;
13
14use backoff::{BackoffError, BackoffSchedule, RetriableError, Runner};
15use descriptor::{DescriptorStatus, VersionedDescriptor, build_sign};
16use reactor::Reactor;
17use reactor::read_blind_id_keypair;
18use reupload_timer::ReuploadTimer;
19
20use tor_config_path::CfgPathResolver;
21
22pub use reactor::UploadError;
23pub(crate) use reactor::{Mockable, OVERALL_UPLOAD_TIMEOUT, Real};
24
25/// A handle for the Hsdir Publisher for an onion service.
26///
27/// This handle represents a set of tasks that identify the hsdirs for each
28/// relevant time period, construct descriptors, publish them, and keep them
29/// up-to-date.
30#[must_use = "If you don't call launch() on the publisher, it won't publish any descriptors."]
31pub(crate) struct Publisher<R: Runtime, M: Mockable> {
32    /// The runtime.
33    runtime: R,
34    /// The service for which we're publishing descriptors.
35    nickname: HsNickname,
36    /// A source for new network directories that we use to determine
37    /// our HsDirs.
38    dir_provider: Arc<dyn NetDirProvider>,
39    /// Mockable state.
40    ///
41    /// This is used for launching circuits and for obtaining random number generators.
42    mockable: M,
43    /// The onion service config.
44    config: Arc<OnionServiceConfig>,
45    /// A channel for receiving IPT change notifications.
46    ipt_watcher: IptsPublisherView,
47    /// A channel for receiving onion service config change notifications.
48    config_rx: watch::Receiver<Arc<OnionServiceConfig>>,
49    /// The key manager.
50    keymgr: Arc<KeyMgr>,
51    /// A sender for updating the status of the onion service.
52    status_tx: PublisherStatusSender,
53    /// Path resolver for configuration files.
54    path_resolver: Arc<CfgPathResolver>,
55    /// Proof-of-work state
56    pow_manager: Arc<PowManager<R>>,
57    /// Queue on which we receive messages from the [`PowManager`] telling us that a seed has
58    /// rotated and thus we need to republish the descriptor for a particular time period.
59    update_from_pow_manager_rx: mpsc::Receiver<TimePeriod>,
60}
61
62impl<R: Runtime, M: Mockable> Publisher<R, M> {
63    /// Create a new publisher.
64    ///
65    /// When it launches, it will know no keys or introduction points,
66    /// and will therefore not upload any descriptors.
67    ///
68    /// The publisher won't start publishing until you call [`Publisher::launch`].
69    #[allow(clippy::too_many_arguments)]
70    pub(crate) fn new(
71        runtime: R,
72        nickname: HsNickname,
73        dir_provider: Arc<dyn NetDirProvider>,
74        mockable: impl Into<M>,
75        ipt_watcher: IptsPublisherView,
76        config_rx: watch::Receiver<Arc<OnionServiceConfig>>,
77        status_tx: PublisherStatusSender,
78        keymgr: Arc<KeyMgr>,
79        path_resolver: Arc<CfgPathResolver>,
80        pow_manager: Arc<PowManager<R>>,
81        update_from_pow_manager_rx: mpsc::Receiver<TimePeriod>,
82    ) -> Self {
83        let config = config_rx.borrow().clone();
84        Self {
85            runtime,
86            nickname,
87            dir_provider,
88            mockable: mockable.into(),
89            config,
90            ipt_watcher,
91            config_rx,
92            status_tx,
93            keymgr,
94            path_resolver,
95            pow_manager,
96            update_from_pow_manager_rx,
97        }
98    }
99
100    /// Launch the publisher reactor.
101    pub(crate) fn launch(self) -> Result<(), StartupError> {
102        let Publisher {
103            runtime,
104            nickname,
105            dir_provider,
106            mockable,
107            config,
108            ipt_watcher,
109            config_rx,
110            status_tx,
111            keymgr,
112            path_resolver,
113            pow_manager,
114            update_from_pow_manager_rx: publisher_update_rx,
115        } = self;
116
117        let reactor = Reactor::new(
118            runtime.clone(),
119            nickname,
120            dir_provider,
121            mockable,
122            &config,
123            ipt_watcher,
124            config_rx,
125            status_tx,
126            keymgr,
127            path_resolver,
128            pow_manager,
129            publisher_update_rx,
130        );
131
132        runtime
133            .spawn(async move {
134                match reactor.run().await {
135                    Ok(()) => debug!("the publisher reactor has shut down"),
136                    Err(e) => warn_report!(e, "the publisher reactor has shut down"),
137                }
138            })
139            .map_err(|e| StartupError::Spawn {
140                spawning: "publisher reactor task",
141                cause: e.into(),
142            })?;
143
144        Ok(())
145    }
146}
147
148#[cfg(test)]
149mod test {
150    // @@ begin test lint list maintained by maint/add_warning @@
151    #![allow(clippy::bool_assert_comparison)]
152    #![allow(clippy::clone_on_copy)]
153    #![allow(clippy::dbg_macro)]
154    #![allow(clippy::mixed_attributes_style)]
155    #![allow(clippy::print_stderr)]
156    #![allow(clippy::print_stdout)]
157    #![allow(clippy::single_char_pattern)]
158    #![allow(clippy::unwrap_used)]
159    #![allow(clippy::unchecked_time_subtraction)]
160    #![allow(clippy::useless_vec)]
161    #![allow(clippy::needless_pass_by_value)]
162    #![allow(clippy::string_slice)] // See arti#2571
163    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
164    use super::*;
165
166    use std::collections::HashMap;
167    use std::io;
168    use std::path::Path;
169    use std::pin::Pin;
170    use std::sync::Mutex;
171    use std::sync::atomic::{AtomicUsize, Ordering};
172    use std::task::{Context, Poll};
173
174    use async_trait::async_trait;
175    use fs_mistrust::Mistrust;
176    use futures::{AsyncRead, AsyncWrite};
177    use tempfile::{TempDir, tempdir};
178    use test_temp_dir::test_temp_dir;
179
180    use tor_basic_utils::test_rng::{TestingRng, testing_rng};
181
182    use tor_hscrypto::pk::{HsBlindId, HsDescSigningKeypair, HsId, HsIdKey, HsIdKeypair};
183    use tor_key_forge::ToEncodableKey;
184    use tor_keymgr::{ArtiNativeKeystore, KeyMgrBuilder, KeySpecifier};
185    use tor_llcrypto::pk::{ed25519, rsa};
186    use tor_netdir::testprovider::TestNetDirProvider;
187    use tor_netdir::{NetDir, testnet};
188    use tor_netdoc::doc::hsdesc::test_data;
189    use tor_rtcompat::ToplevelBlockOn;
190    use tor_rtmock::MockRuntime;
191
192    use crate::HsNickname;
193    use crate::config::OnionServiceConfigBuilder;
194    use crate::ipt_set::{IptInSet, IptSet, ipts_channel};
195    use crate::pow::NewPowManager;
196    use crate::publish::reactor::MockableDirTunnel;
197    use crate::status::{OnionServiceStatus, StatusSender};
198    use crate::test::create_storage_handles;
199    use crate::{
200        BlindIdKeypairSpecifier, BlindIdPublicKeySpecifier, DescSigningKeypairSpecifier,
201        HsIdKeypairSpecifier, HsIdPublicKeySpecifier,
202    };
203
204    /// The nickname of the test service.
205    const TEST_SVC_NICKNAME: &str = "test-svc";
206
207    /// The HTTP response the HSDir returns if everything went well.
208    const OK_RESPONSE: &str = "HTTP/1.1 200 OK\r\n\r\n";
209
210    /// The HTTP response the HSDir returns if something went wrong
211    const ERR_RESPONSE: &str = "HTTP/1.1 500 UH_OH\r\n\r\n";
212
213    /// The error doesn't matter (we return a dummy io::Error from poll_read).
214    ///
215    /// NOTE: ideally, this would be an io::Result, but io::Error isn't Clone (the tests need to
216    /// clone the iterator over these Results for each HSDir).
217    type PollReadResult<T> = Result<T, ()>;
218
219    /// A trait for our poll_read response iterator.
220    trait PollReadIter:
221        Iterator<Item = PollReadResult<String>> + Send + Sync + Clone + Unpin + 'static
222    {
223    }
224
225    impl<I> PollReadIter for I where
226        I: Iterator<Item = PollReadResult<String>> + Send + Sync + Clone + Unpin + 'static
227    {
228    }
229
230    #[derive(Clone, Debug, Default)]
231    struct MockReactorState<I: PollReadIter> {
232        /// The number of `POST /tor/hs/3/publish` requests sent by the reactor.
233        publish_count: Arc<AtomicUsize>,
234        /// The values returned by `DataStream::poll_read` when uploading to an HSDir.
235        ///
236        /// The values represent the HTTP response (or lack thereof) each HSDir sends upon
237        /// receiving a POST request for uploading a descriptor.
238        ///
239        /// Note: this field is only used for populating responses_for_hsdir. Each time
240        /// get_or_launch_specific is called for a new CircTarget, this iterator is cloned and
241        /// added to the responses_for_hsdir entry corresponding to the new CircTarget (HSDir).
242        poll_read_responses: I,
243        /// The responses that will be returned by each test HSDir (identified by its RsaIdentity).
244        ///
245        /// Used for testing whether the reactor correctly retries on failure.
246        responses_for_hsdir: Arc<Mutex<HashMap<rsa::RsaIdentity, I>>>,
247    }
248
249    #[async_trait]
250    impl<I: PollReadIter> Mockable for MockReactorState<I> {
251        type Rng = TestingRng;
252        type Tunnel = MockClientCirc<I>;
253
254        fn thread_rng(&self) -> Self::Rng {
255            testing_rng()
256        }
257
258        async fn get_or_launch_hs_dir<T>(
259            &self,
260            _netdir: &tor_netdir::NetDir,
261            target: T,
262        ) -> Result<Self::Tunnel, tor_circmgr::Error>
263        where
264            T: tor_linkspec::CircTarget + Send + Sync,
265        {
266            // Look up the next poll_read value to return for this relay.
267            let id = target.rsa_identity().unwrap();
268            let mut map = self.responses_for_hsdir.lock().unwrap();
269            let poll_read_responses = map
270                .entry(*id)
271                .or_insert_with(|| self.poll_read_responses.clone());
272
273            Ok(MockClientCirc {
274                publish_count: Arc::clone(&self.publish_count),
275                poll_read_responses: poll_read_responses.clone(),
276            })
277        }
278
279        fn estimate_upload_timeout(&self) -> Duration {
280            // chosen arbitrarily for testing.
281            Duration::from_secs(30)
282        }
283    }
284
285    #[derive(Debug, Clone)]
286    struct MockClientCirc<I: PollReadIter> {
287        /// The number of `POST /tor/hs/3/publish` requests sent by the reactor.
288        publish_count: Arc<AtomicUsize>,
289        /// The values to return from `poll_read`.
290        ///
291        /// Used for testing whether the reactor correctly retries on failure.
292        poll_read_responses: I,
293    }
294
295    #[async_trait]
296    impl<I: PollReadIter> MockableDirTunnel for MockClientCirc<I> {
297        type DataStream = MockDataStream<I>;
298
299        async fn begin_dir_stream(&self) -> Result<Self::DataStream, tor_circmgr::Error> {
300            Ok(MockDataStream {
301                publish_count: Arc::clone(&self.publish_count),
302                // TODO: this will need to change when we start reusing circuits (currently,
303                // we only ever create one data stream per circuit).
304                poll_read_responses: self.poll_read_responses.clone(),
305                at_start: true,
306            })
307        }
308
309        fn source_info(&self) -> tor_proto::Result<Option<tor_dirclient::SourceInfo>> {
310            Ok(None)
311        }
312    }
313
314    #[derive(Debug)]
315    struct MockDataStream<I: PollReadIter> {
316        /// The number of `POST /tor/hs/3/publish` requests sent by the reactor.
317        publish_count: Arc<AtomicUsize>,
318        /// The values to return from `poll_read`.
319        ///
320        /// Used for testing whether the reactor correctly retries on failure.
321        poll_read_responses: I,
322        /// Are we at the start of a stream?
323        ///
324        /// (We keep track of this so we can assert that streams begin with a reasonable
325        /// HTTP header.)
326        at_start: bool,
327    }
328
329    impl<I: PollReadIter> AsyncRead for MockDataStream<I> {
330        fn poll_read(
331            mut self: Pin<&mut Self>,
332            _cx: &mut Context<'_>,
333            buf: &mut [u8],
334        ) -> Poll<io::Result<usize>> {
335            match self.as_mut().poll_read_responses.next() {
336                Some(res) => {
337                    match res {
338                        Ok(res) => {
339                            buf[..res.len()].copy_from_slice(res.as_bytes());
340
341                            Poll::Ready(Ok(res.len()))
342                        }
343                        Err(()) => {
344                            // Return an error. This should cause the reactor to reattempt the
345                            // upload.
346                            Poll::Ready(Err(io::Error::other("test error")))
347                        }
348                    }
349                }
350                None => Poll::Ready(Ok(0)),
351            }
352        }
353    }
354
355    impl<I: PollReadIter> AsyncWrite for MockDataStream<I> {
356        fn poll_write(
357            mut self: Pin<&mut Self>,
358            _cx: &mut Context<'_>,
359            buf: &[u8],
360        ) -> Poll<io::Result<usize>> {
361            let request = std::str::from_utf8(buf).unwrap();
362
363            if self.at_start {
364                assert!(request.starts_with("POST /tor/hs/3/publish HTTP/1.0\r\n"));
365                let _prev = self.publish_count.fetch_add(1, Ordering::SeqCst);
366                self.as_mut().at_start = false;
367            }
368
369            Poll::Ready(Ok(request.len()))
370        }
371
372        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
373            Poll::Ready(Ok(()))
374        }
375
376        fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
377            Poll::Ready(Ok(()))
378        }
379    }
380
381    /// Insert the specified key into the keystore.
382    fn insert_svc_key<K>(key: K, keymgr: &KeyMgr, svc_key_spec: &dyn KeySpecifier)
383    where
384        K: ToEncodableKey,
385    {
386        keymgr
387            .insert(
388                key,
389                svc_key_spec,
390                tor_keymgr::KeystoreSelector::Primary,
391                true,
392            )
393            .unwrap();
394    }
395
396    /// Create a new `KeyMgr`, provisioning its keystore with the necessary keys.
397    fn init_keymgr(
398        keystore_dir: &TempDir,
399        nickname: &HsNickname,
400        netdir: &NetDir,
401    ) -> (HsId, HsBlindId, Arc<KeyMgr>) {
402        let period = netdir.hs_time_period();
403
404        let mut rng = testing_rng();
405        let keypair = ed25519::Keypair::generate(&mut rng);
406        let id_pub = HsIdKey::from(keypair.verifying_key());
407        let id_keypair = HsIdKeypair::from(ed25519::ExpandedKeypair::from(&keypair));
408
409        let (hs_blind_id_key, hs_blind_id_kp, _subcredential) =
410            id_keypair.compute_blinded_key(period).unwrap();
411
412        let keystore = ArtiNativeKeystore::from_path_and_mistrust(
413            keystore_dir,
414            &Mistrust::new_dangerously_trust_everyone(),
415        )
416        .unwrap();
417
418        // Provision the keystore with the necessary keys:
419        let keymgr = KeyMgrBuilder::default()
420            .primary_store(Box::new(keystore))
421            .build()
422            .unwrap();
423
424        insert_svc_key(
425            id_keypair,
426            &keymgr,
427            &HsIdKeypairSpecifier::new(nickname.clone()),
428        );
429
430        insert_svc_key(
431            id_pub.clone(),
432            &keymgr,
433            &HsIdPublicKeySpecifier::new(nickname.clone()),
434        );
435
436        insert_svc_key(
437            hs_blind_id_kp,
438            &keymgr,
439            &BlindIdKeypairSpecifier::new(nickname.clone(), period),
440        );
441
442        insert_svc_key(
443            hs_blind_id_key.clone(),
444            &keymgr,
445            &BlindIdPublicKeySpecifier::new(nickname.clone(), period),
446        );
447
448        insert_svc_key(
449            HsDescSigningKeypair::from(ed25519::Keypair::generate(&mut rng)),
450            &keymgr,
451            &DescSigningKeypairSpecifier::new(nickname.clone(), period),
452        );
453
454        let hs_id = id_pub.into();
455        (hs_id, hs_blind_id_key.into(), keymgr.into())
456    }
457
458    fn build_test_config(nickname: HsNickname) -> OnionServiceConfig {
459        OnionServiceConfigBuilder::default()
460            .nickname(nickname)
461            .rate_limit_at_intro(None)
462            .build()
463            .unwrap()
464    }
465
466    #[allow(clippy::too_many_arguments)]
467    fn run_test<I: PollReadIter>(
468        runtime: MockRuntime,
469        nickname: HsNickname,
470        keymgr: Arc<KeyMgr>,
471        pv: IptsPublisherView,
472        config_rx: watch::Receiver<Arc<OnionServiceConfig>>,
473        status_tx: StatusSender,
474        netdir: NetDir,
475        reactor_event: impl FnOnce(),
476        poll_read_responses: I,
477        expected_upload_count: usize,
478        republish_count: usize,
479        expect_errors: bool,
480    ) {
481        runtime.clone().block_on(async move {
482            let netdir_provider: Arc<dyn NetDirProvider> =
483                Arc::new(TestNetDirProvider::from(netdir));
484            let publish_count = Default::default();
485            let circpool = MockReactorState {
486                publish_count: Arc::clone(&publish_count),
487                poll_read_responses,
488                responses_for_hsdir: Arc::new(Mutex::new(Default::default())),
489            };
490
491            let temp_dir = test_temp_dir!();
492            let state_dir = temp_dir.subdir_untracked("state_dir");
493            let mistrust = fs_mistrust::Mistrust::new_dangerously_trust_everyone();
494            let state_dir = StateDirectory::new(state_dir, &mistrust).unwrap();
495            let state_handle = state_dir.acquire_instance(&nickname).unwrap();
496            let pow_nonce_dir = state_handle.raw_subdir("pow_nonces").unwrap();
497            let pow_manager_storage_handle = state_handle.storage_handle("pow_manager").unwrap();
498
499            let NewPowManager {
500                pow_manager,
501                rend_req_tx: _,
502                rend_req_rx: _,
503                publisher_update_rx: update_from_pow_manager_rx,
504            } = PowManager::new(
505                runtime.clone(),
506                nickname.clone(),
507                pow_nonce_dir,
508                keymgr.clone(),
509                pow_manager_storage_handle,
510                netdir_provider.clone(),
511                status_tx.clone().into(),
512                config_rx.clone(),
513            )
514            .unwrap();
515            let mut status_rx = status_tx.subscribe();
516            let publisher: Publisher<MockRuntime, MockReactorState<_>> = Publisher::new(
517                runtime.clone(),
518                nickname,
519                netdir_provider,
520                circpool,
521                pv,
522                config_rx,
523                status_tx.into(),
524                keymgr,
525                Arc::new(CfgPathResolver::default()),
526                pow_manager,
527                update_from_pow_manager_rx,
528            );
529
530            publisher.launch().unwrap();
531            runtime.progress_until_stalled().await;
532            let status = status_rx.next().await.unwrap().publisher_status();
533            assert_eq!(State::Shutdown, status.state());
534            assert!(status.current_problem().is_none());
535
536            // Check that we haven't published anything yet
537            assert_eq!(publish_count.load(Ordering::SeqCst), 0);
538
539            reactor_event();
540
541            runtime.progress_until_stalled().await;
542
543            // We need to manually advance the time, because some of our tests check that the
544            // failed uploads are retried, and there's a sleep() between the retries
545            // (see BackoffSchedule::next_delay).
546            runtime.advance_by(Duration::from_secs(1)).await;
547            runtime.progress_until_stalled().await;
548
549            let initial_publish_count = publish_count.load(Ordering::SeqCst);
550            assert_eq!(initial_publish_count, expected_upload_count);
551
552            let status = status_rx.next().await.unwrap().publisher_status();
553            if expect_errors {
554                // The upload results aren't ready yet.
555                assert_eq!(State::Bootstrapping, status.state());
556            } else {
557                // The test network doesn't have an SRV for the previous TP,
558                // so we are "unreachable".
559                assert_eq!(State::DegradedUnreachable, status.state());
560            }
561            assert!(status.current_problem().is_none());
562
563            if republish_count > 0 {
564                /// The latest time the descriptor can be republished.
565                const MAX_TIMEOUT: Duration = Duration::from_secs(60 * 120);
566
567                // Wait until the reactor triggers the necessary number of reuploads.
568                runtime
569                    .advance_by(MAX_TIMEOUT * (republish_count as u32))
570                    .await;
571                runtime.progress_until_stalled().await;
572
573                let min_upload_count = expected_upload_count * republish_count;
574                // There will be twice as many reuploads if the publisher happens
575                // to reupload every hour (as opposed to every 2h).
576                let max_upload_count = 2 * min_upload_count;
577                let publish_count_now = publish_count.load(Ordering::SeqCst);
578                // This is the total number of reuploads (i.e. the number of times
579                // we published the descriptor to an HsDir).
580                let actual_reupload_count = publish_count_now - initial_publish_count;
581
582                assert!((min_upload_count..=max_upload_count).contains(&actual_reupload_count));
583            }
584        });
585    }
586
587    /// Test that the publisher publishes the descriptor when the IPTs change.
588    ///
589    /// The `poll_read_responses` are returned by each HSDir, in order, in response to each POST
590    /// request received from the publisher.
591    ///
592    /// The `multiplier` represents the multiplier by which to multiply the number of HSDirs to
593    /// obtain the total expected number of uploads (this works because the test "HSDirs" all
594    /// behave the same, so the number of uploads is the number of HSDirs multiplied by the number
595    /// of retries).
596    fn publish_after_ipt_change<I: PollReadIter>(
597        temp_dir: &Path,
598        poll_read_responses: I,
599        multiplier: usize,
600        republish_count: usize,
601        expect_errors: bool,
602    ) {
603        let runtime = MockRuntime::new();
604        let nickname = HsNickname::try_from(TEST_SVC_NICKNAME.to_string()).unwrap();
605        let config = build_test_config(nickname.clone());
606        let (_config_tx, config_rx) = watch::channel_with(Arc::new(config));
607
608        let (mut mv, pv) = ipts_channel(&runtime, create_storage_handles(temp_dir).1).unwrap();
609        let update_ipts = || {
610            let ipts: Vec<IptInSet> = test_data::test_parsed_hsdesc()
611                .unwrap()
612                .intro_points()
613                .iter()
614                .enumerate()
615                .map(|(i, ipt)| IptInSet {
616                    ipt: ipt.clone(),
617                    lid: [i.try_into().unwrap(); 32].into(),
618                })
619                .collect();
620
621            mv.borrow_for_update(runtime.clone()).ipts = Some(IptSet {
622                ipts,
623                lifetime: Duration::from_secs(20),
624            });
625        };
626
627        let netdir = testnet::construct_netdir().unwrap_if_sufficient().unwrap();
628        let keystore_dir = tempdir().unwrap();
629
630        let (_hsid, blind_id, keymgr) = init_keymgr(&keystore_dir, &nickname, &netdir);
631
632        let hsdir_count = netdir
633            .hs_dirs_upload(blind_id, netdir.hs_time_period())
634            .unwrap()
635            .collect::<Vec<_>>()
636            .len();
637
638        assert!(hsdir_count > 0);
639
640        // If any of the uploads fail, they will be retried. Note that the upload failure will
641        // affect _each_ hsdir, so the expected number of uploads is a multiple of hsdir_count.
642        let expected_upload_count = hsdir_count * multiplier;
643        let status_tx = StatusSender::new(OnionServiceStatus::new_shutdown());
644
645        run_test(
646            runtime.clone(),
647            nickname,
648            keymgr,
649            pv,
650            config_rx,
651            status_tx,
652            netdir,
653            update_ipts,
654            poll_read_responses,
655            expected_upload_count,
656            republish_count,
657            expect_errors,
658        );
659    }
660
661    #[test]
662    fn publish_after_ipt_change_no_errors() {
663        // The HSDirs always respond with 200 OK, so we expect to publish hsdir_count times.
664        let poll_reads = [Ok(OK_RESPONSE.into())].into_iter();
665
666        test_temp_dir!().used_by(|dir| publish_after_ipt_change(dir, poll_reads, 1, 0, false));
667    }
668
669    #[test]
670    fn publish_after_ipt_change_with_errors() {
671        let err_responses = vec![
672            // The HSDir closed the connection without sending a response.
673            Err(()),
674            // The HSDir responded with an internal server error,
675            Ok(ERR_RESPONSE.to_string()),
676        ];
677
678        for error_res in err_responses.into_iter() {
679            let poll_reads = vec![
680                // Each HSDir first responds with an error, which causes the publisher to retry the
681                // upload. The HSDir then responds with "200 OK".
682                //
683                // We expect to publish hsdir_count * 2 times (for each HSDir, the first upload
684                // attempt fails, but the second succeeds).
685                error_res,
686                Ok(OK_RESPONSE.to_string()),
687            ]
688            .into_iter();
689
690            test_temp_dir!().used_by(|dir| publish_after_ipt_change(dir, poll_reads, 2, 0, true));
691        }
692    }
693
694    #[test]
695    fn reupload_after_publishing() {
696        let poll_reads = [Ok(OK_RESPONSE.into())].into_iter();
697        // Test that 4 reuploads happen after the initial upload
698        const REUPLOAD_COUNT: usize = 4;
699
700        test_temp_dir!()
701            .used_by(|dir| publish_after_ipt_change(dir, poll_reads, 1, REUPLOAD_COUNT, false));
702    }
703
704    // TODO (#1120): test that the descriptor is republished when the config changes
705
706    // TODO (#1120): test that the descriptor is reuploaded only to the HSDirs that need it (i.e. the
707    // ones for which it's dirty)
708
709    // TODO (#1120): test that rate-limiting works correctly
710
711    // TODO (#1120): test that the uploaded descriptor contains the expected values
712
713    // TODO (#1120): test that the publisher stops publishing if the IPT manager sets the IPTs to
714    // `None`.
715}
tor_hsservice/publish.rs

tor_hsservice/
publish.rs
