//! Main implementation of the connection functionality use std::collections::HashMap; use std::fmt::Debug; use std::marker::PhantomData; use std::sync::Arc; use async_trait::async_trait; use educe::Educe; use futures::{AsyncRead, AsyncWrite}; use itertools::Itertools; use rand::RngExt; use tor_bytes::Writeable; use tor_cell::relaycell::hs::IntroduceAckStatus; use tor_cell::relaycell::hs::intro_payload::{self, IntroduceHandshakePayload}; use tor_cell::relaycell::hs::pow::ProofOfWork; use tor_cell::relaycell::msg::{AnyRelayMsg, Introduce1, Rendezvous2}; use tor_circmgr::build::onion_circparams_from_netparams; use tor_circmgr::{ ClientOnionServiceDataTunnel, ClientOnionServiceDirTunnel, ClientOnionServiceIntroTunnel, }; use tor_dirclient::SourceInfo; use tor_error::{Bug, debug_report, warn_report}; use tor_hscrypto::Subcredential; use tor_hscrypto::time::TimePeriod; use tor_netdir::params::NetParameters; use tor_proto::TargetHop; use tor_proto::client::circuit::handshake::hs_ntor::{self, HsNtorHkdfKeyGenerator}; use tracing::{debug, instrument, trace, warn}; use web_time_compat::{Duration, Instant, SystemTime}; use retry_error::RetryError; use safelog::{DispRedacted, Sensitive}; use tor_cell::relaycell::RelayMsg; use tor_cell::relaycell::hs::{ AuthKeyType, EstablishRendezvous, IntroduceAck, RendezvousEstablished, }; use tor_checkable::{TimeBound, timed::TimeRangeBound}; use tor_circmgr::hspool::HsCircPool; use tor_circmgr::timeouts::Action as TimeoutsAction; use tor_dirclient::request::Requestable as _; use tor_error::{HasRetryTime as _, RetryTime}; use tor_error::{internal, into_internal}; use tor_hscrypto::RendCookie; use tor_hscrypto::pk::{HsBlindId, HsId, HsIdKey}; use tor_linkspec::{CircTarget, HasRelayIds, OwnedCircTarget, RelayId}; use tor_llcrypto::pk::ed25519::Ed25519Identity; use tor_netdir::{NetDir, Relay}; use tor_netdoc::doc::hsdesc::{HsDesc, IntroPointDesc}; use tor_proto::client::circuit::{CircParameters, handshake}; use tor_proto::{MetaCellDisposition, MsgHandler}; use tor_rtcompat::{Runtime, SleepProviderExt as _, TimeoutError}; use crate::Config; use crate::caps; use crate::err::RendPtIdentityForError; use crate::pow::HsPowClient; use crate::proto_oneshot; use crate::relay_info::ipt_to_circtarget; use crate::state::MockableConnectorData; use crate::{ConnError, DescriptorError, DescriptorErrorDetail}; use crate::{FailedAttemptError, IntroPtIndex, rend_pt_identity_for_error}; use crate::{HsClientConnector, HsClientSecretKeys}; use ConnError as CE; use FailedAttemptError as FAE; /// Given `R, M` where `M: MocksForConnect`, expand to the mockable `ClientCirc` // This is quite annoying. But the alternative is to write out `<... as // ...>` // each time, since otherwise the compile complains about ambiguous associated types. macro_rules! DataTunnel{ { $R:ty, $M:ty } => { <<$M as MocksForConnect<$R>>::HsCircPool as MockableCircPool<$R>>::DataTunnel } } /// Information about a hidden service, including our connection history #[derive(Default, Educe)] #[educe(Debug)] // This type is actually crate-private, since it isn't re-exported, but it must // be `pub` because it appears as a default for a type parameter in HsClientConnector. pub struct Data { /// The latest known onion service descriptor for this service. desc: DataHsDesc, /// Information about the latest status of trying to connect to this service /// through each of its introduction points. ipts: DataIpts, /// Information about the requery period of each HsDir we have recently queried. /// /// Each entry represents an HsDir that we cannot requery until /// its specified timestamp elapses. /// /// Any HsDir that does not have an entry in this map can be requeried. hsdirs: DataHsDirs, } /// An onion service descriptor and its associated HsBlindId. #[derive(Debug)] struct HsDescForTp { /// The TP this descriptor is for. /// /// Used for determining whether a newly fetched descriptor /// is for the same time period as this one. time_period: TimePeriod, /// The descriptor desc: TimeRangeBound, } /// Part of `Data` that relates to our information about the HsDir requery periods type DataHsDirs = HashMap; /// Marker type, to make typed HsDir [`RelayIdFor`] keys #[derive(Hash, Eq, PartialEq, Ord, PartialOrd, Copy, Clone, Debug)] struct RequeryPeriodMap; /// Lookup key for looking up and recording our IPT use experiences type RelayIdForRequeryPeriod = RelayIdFor; /// Part of `Data` that relates to the HS descriptor type DataHsDesc = Option; /// Part of `Data` that relates to our information about introduction points type DataIpts = HashMap; /// How things went last time we tried to use this introduction point /// /// Neither this data structure, nor [`Data`], is responsible for arranging that we expire this /// information eventually. If we keep reconnecting to the service, we'll retain information /// about each IPT indefinitely, at least so long as they remain listed in the descriptors we /// receive. /// /// Expiry of unused data is handled by `state.rs`, according to `last_used` in `ServiceState`. /// /// Choosing which IPT to prefer is done by obtaining an `IptSortKey` /// (from this and other information). // // Don't impl Ord for IptExperience. We obtain `Option<&IptExperience>` from our // data structure, and if IptExperience were Ord then Option<&IptExperience> would be Ord // but it would be the wrong sort order: it would always prefer None, ie untried IPTs. #[derive(Debug)] struct IptExperience { /// How long it took us to get whatever outcome occurred /// /// We prefer fast successes to slow ones. /// Then, we prefer failures with earlier `RetryTime`, /// and, lastly, faster failures to slower ones. duration: Duration, /// What happened and when we might try again /// /// Note that we don't actually *enforce* the `RetryTime` here, just sort by it /// using `RetryTime::loose_cmp`. /// /// We *do* return an error that is itself `HasRetryTime` and expect our callers /// to honour that. outcome: Result<(), RetryTime>, } /// Actually make a HS connection, updating our recorded state as necessary /// /// `connector` is provided only for obtaining the runtime and netdir (and `mock_for_state`). /// Obviously, `connect` is not supposed to go looking in `services`. /// /// This function handles all necessary retrying of fallible operations, /// (and, therefore, must also limit the total work done for a particular call). /// /// This function has a minimum of functionality, since it is the boundary /// between "mock connection, used for testing `state.rs`" and /// "mock circuit and netdir, used for testing `connect.rs`", /// so it is not, itself, unit-testable. #[instrument(level = "trace", skip_all)] pub(crate) async fn connect( connector: &HsClientConnector, netdir: Arc, config: Arc, hsid: HsId, data: &mut Data, secret_keys: HsClientSecretKeys, ) -> Result { Context::new( &connector.runtime, &*connector.circpool, netdir, config, hsid, secret_keys, (), )? .connect(data) .await } /// Common context for a single request to connect to a hidden service /// /// This saves on passing this same set of (immutable) values (or subsets thereof) /// to each method in the principal functional code, everywhere. /// It also provides a convenient type to be `Self`. /// /// Its lifetime is one request to make a new client circuit to a hidden service, /// including all the retries and timeouts. struct Context<'c, R: Runtime, M: MocksForConnect> { /// Runtime runtime: &'c R, /// Circpool circpool: &'c M::HsCircPool, /// Netdir // // TODO holding onto the netdir for the duration of our attempts is not ideal // but doing better is fairly complicated. See discussions here: // https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1228#note_2910545 // https://gitlab.torproject.org/tpo/core/arti/-/issues/884 netdir: Arc, /// Configuration config: Arc, /// Secret keys to use secret_keys: HsClientSecretKeys, /// HS ID hsid: DispRedacted, /// Blinded HS ID hs_blind_id: HsBlindId, /// The subcredential to use during this time period subcredential: Subcredential, /// Mock data mocks: M, } /// Details of an established rendezvous point /// /// Intermediate value for progress during a connection attempt. struct Rendezvous<'r, R: Runtime, M: MocksForConnect> { /// RPT as a `Relay` rend_relay: Relay<'r>, /// Rendezvous circuit rend_tunnel: DataTunnel!(R, M), /// Rendezvous cookie rend_cookie: RendCookie, /// Receiver that will give us the RENDEZVOUS2 message. /// /// The sending ended is owned by the handler /// which receives control messages on the rendezvous circuit, /// and which was installed when we sent `ESTABLISH_RENDEZVOUS`. /// /// (`RENDEZVOUS2` is the message containing the onion service's side of the handshake.) rend2_rx: proto_oneshot::Receiver, /// Dummy, to placate compiler /// /// Covariant without dropck or interfering with Send/Sync will do fine. marker: PhantomData (R, M)>, } /// Random value used as part of IPT selection type IptSortRand = u32; /// Details of an apparently-useable introduction point /// /// Intermediate value for progress during a connection attempt. struct UsableIntroPt<'i> { /// Index in HS descriptor intro_index: IntroPtIndex, /// IPT descriptor intro_desc: &'i IntroPointDesc, /// IPT `CircTarget` intro_target: OwnedCircTarget, /// Random value used as part of IPT selection sort_rand: IptSortRand, } /// Lookup key for looking up and recording information about a relay /// /// Used to identify a relay when looking to see what happened last time we used it, /// and storing that information after we tried it. /// /// We store the experience information under an arbitrary one of the relay's identities, /// as returned by the `HasRelayIds::identities().next()`. /// When we do lookups, we check all the relay's identities to see if we find /// anything relevant. /// If relay identities permute in strange ways, whether we find our previous /// knowledge about them is not particularly well defined, but that's fine. /// /// While this is, structurally, a relay identity, it is not suitable for other purposes. #[derive(Hash, Eq, PartialEq, Ord, PartialOrd, Debug)] struct RelayIdFor { /// The relay id inner: RelayId, /// Phantom data to allow parameterizing over `K` /// /// `K` is a marker type that represents the kind of map /// this key will be used in. marker: PhantomData, } /// Marker type, to make typed Ipt exprience [`RelayIdFor`] keys #[derive(Hash, Eq, PartialEq, Ord, PartialOrd, Copy, Clone, Debug)] struct IptExperienceMap; /// Lookup key for looking up and recording our IPT use experiences type RelayIdForExperience = RelayIdFor; /// Details of an apparently-successful INTRODUCE exchange /// /// Intermediate value for progress during a connection attempt. struct Introduced> { /// End-to-end crypto NTORv3 handshake with the service /// /// Created as part of generating our `INTRODUCE1`, /// and then used when processing `RENDEZVOUS2`. handshake_state: hs_ntor::HsNtorClientState, /// A set of peer extensions that we decided to negotiate and use. peer_caps: caps::PeerCaps, /// Dummy, to placate compiler /// /// `R` and `M` only used for getting to mocks. /// Covariant without dropck or interfering with Send/Sync will do fine. marker: PhantomData (R, M)>, } impl RelayIdFor { /// Create a new key for use with `T` fn new(inner: RelayId) -> Self { Self { inner, marker: Default::default(), } } /// Identities to use to try to find previous experience information about this IPT fn for_lookup(ids: &T) -> impl Iterator + '_ { ids.identities().map(|id| RelayIdFor::new(id.to_owned())) } /// Identity to use to store previous experience information about this IPT fn for_store(ids: &T) -> Result { let id = ids .identities() .next() .ok_or_else(|| internal!("introduction point relay with no identities"))? .to_owned(); Ok(RelayIdFor::new(id)) } } /// Sort key for an introduction point, for selecting the best IPTs to try first /// /// Ordering is most preferable first. /// /// We use this to sort our `UsableIpt`s using `.sort_by_key`. /// (This implementation approach ensures that we obey all the usual ordering invariants.) #[derive(Ord, PartialOrd, Eq, PartialEq, Debug)] struct IptSortKey { /// Sort by how preferable the experience was outcome: IptSortKeyOutcome, /// Failing that, choose randomly sort_rand: IptSortRand, } /// Component of the [`IptSortKey`] representing outcome of our last attempt, if any /// /// This is the main thing we use to decide which IPTs to try first. /// It is calculated for each IPT /// (via `.sort_by_key`, so repeatedly - it should therefore be cheap to make.) /// /// Ordering is most preferable first. #[derive(Ord, PartialOrd, Eq, PartialEq, Debug)] enum IptSortKeyOutcome { /// Prefer successes Success { /// Prefer quick ones duration: Duration, }, /// Failing that, try one we don't know to have failed Untried, /// Failing that, it'll have to be ones that didn't work last time Failed { /// Prefer failures with an earlier retry time retry_time: tor_error::LooseCmpRetryTime, /// Failing that, prefer quick failures (rather than slow ones eg timeouts) duration: Duration, }, } impl From> for IptSortKeyOutcome { fn from(experience: Option<&IptExperience>) -> IptSortKeyOutcome { use IptSortKeyOutcome as O; match experience { None => O::Untried, Some(IptExperience { duration, outcome }) => match outcome { Ok(()) => O::Success { duration: *duration, }, Err(retry_time) => O::Failed { retry_time: (*retry_time).into(), duration: *duration, }, }, } } } /// Token indicating that a descriptor fetch is wanted #[derive(Clone, Copy, Eq, PartialEq, Debug)] struct RefetchDescriptor; impl<'c, R: Runtime, M: MocksForConnect> Context<'c, R, M> { /// Make a new `Context` from the input data fn new( runtime: &'c R, circpool: &'c M::HsCircPool, netdir: Arc, config: Arc, hsid: HsId, secret_keys: HsClientSecretKeys, mocks: M, ) -> Result { let time_period = netdir.hs_time_period(); let (hs_blind_id_key, subcredential) = HsIdKey::try_from(hsid) .map_err(|_| CE::InvalidHsId)? .compute_blinded_key(time_period) .map_err( // TODO HS what on earth do these errors mean, in practical terms ? // In particular, we'll want to convert them to a ConnError variant, // but what ErrorKind should they have ? into_internal!("key blinding error, don't know how to handle"), )?; let hs_blind_id = hs_blind_id_key.id(); Ok(Context { netdir, config, hsid: DispRedacted(hsid), hs_blind_id, subcredential, circpool, runtime, secret_keys, mocks, }) } /// Actually make a HS connection, updating our recorded state as necessary /// /// Called by the `connect` function in this module. /// /// This function handles all necessary retrying of fallible operations, /// (and, therefore, must also limit the total work done for a particular call). #[instrument(level = "trace", skip_all)] async fn connect(&self, data: &mut Data) -> Result { // This function must do the following, retrying as appropriate. // - Look up the onion descriptor in the state. // - Download the onion descriptor if one isn't there. // - In parallel: // - Pick a rendezvous point from the netdirprovider and launch a // rendezvous circuit to it. Then send ESTABLISH_INTRO. // - Pick a number of introduction points (1 or more) and try to // launch circuits to them. // - On a circuit to an introduction point, send an INTRODUCE1 cell. // - Wait for a RENDEZVOUS2 cell on the rendezvous circuit // - Add a virtual hop to the rendezvous circuit. // - Return the rendezvous circuit. let mocks = self.mocks.clone(); let desc = self .descriptor_ensure(&mut data.desc, &mut data.hsdirs, None) .await?; mocks.test_got_desc(desc); let tunnel = match self.intro_rend_connect(desc, &mut data.ipts).await { Ok(tunnel) => tunnel, Err(e) => { let is_intro_nack = |e| { if let FAE::IntroductionFailed { status, .. } = e { status == IntroduceAckStatus::NOT_RECOGNIZED } else { false } }; let retry = if let CE::Failed(ref errors) = e { // If any of the errors are an INTRODUCE_NACK, // then it's worth retrying one more time // with a fresh descriptor. errors .clone() .into_iter() .any(is_intro_nack) .then_some(RefetchDescriptor) } else { None }; if let Some(RefetchDescriptor) = retry { debug!( "Introduction to {} NACKed, refetching descriptor and retrying", &self.hsid, ); // Refetch the descriptor and try one more time let desc = self .descriptor_ensure(&mut data.desc, &mut data.hsdirs, retry) .await?; mocks.test_got_desc(desc); self.intro_rend_connect(desc, &mut data.ipts).await? } else { return Err(e); } } }; mocks.test_got_tunnel(&tunnel); Ok(tunnel) } /// Ensure that `Data.desc` contains the HS descriptor /// /// If we have a previously-downloaded descriptor, which is still valid, /// just returns a reference to it. /// /// Otherwise, tries to obtain the descriptor by downloading it from hsdir(s). /// /// If `refetch` is `true`, a new descriptor will be refetched /// from the hsdir(s) unconditionally. /// /// Does all necessary retries and timeouts. /// Returns an error if no valid descriptor could be found. #[instrument(level = "trace", skip_all)] async fn descriptor_ensure<'d>( &self, data: &'d mut DataHsDesc, recent_hsdirs: &'d mut DataHsDirs, refetch: Option, ) -> Result<&'d HsDesc, CE> { // Maximum number of hsdir connection and retrieval attempts we'll make let max_total_attempts = self .config .retry .hs_desc_fetch_attempts() .try_into() // User specified a very large u32. We must be downcasting it to 16bit! // let's give them as many retries as we can manage. .unwrap_or(usize::MAX); let now = self.runtime.wallclock(); let unwrap_valid_desc = |data: &'d mut DataHsDesc| -> &'d HsDesc { data.as_ref() .expect("Some but now None") .desc .as_ref() .if_valid_at(&now) .expect("Ok but now Err") }; // We retain a previously obtained descriptor precisely until its lifetime expires, // or until we refetch a more recent one // as a result of an `intro_rend_connect()` failure caused by introduce NACK. // // When it expires, we discard it completely and try to obtain a new one. // // We only replace our cached descriptor if the new one has a higher revision counter. // // TODO SPEC: Discuss HS descriptor lifetime and expiry client behaviour let now = self.runtime.wallclock(); let stored_revision = data.as_ref().and_then(|previously| { if let Ok(desc) = previously.desc.as_ref().if_valid_at(&now) { // Ideally we would just return desc but that confuses borrowck, // so we have to use unwrap_valid_desc() each time // we need the known-to-be-Some descriptor instead. // // https://github.com/rust-lang/rust/issues/51545 Some((desc.revision(), previously.time_period)) } else { // Seems to be not valid now. Try to fetch a fresh one. None } }); match (stored_revision, refetch) { (Some(_), None) => { // Our cached descriptor is still timely, // and we don't need to fetch a new one. return Ok(unwrap_valid_desc(data)); } (None, _) => { // We don't have a timely descriptor, // so ignore the requery_interval, // and reach out to all HsDirs recent_hsdirs.clear(); } (_, Some(RefetchDescriptor)) => { // We have been asked to try to fetch a new descriptor. // We will only reach out to the HsDirs that are // not within the `hs_dir_requery_interval` } } // First, filter out any HsDirs that we *can* requery recent_hsdirs.retain(|_hsdir, requery| *requery > now); let working_tp = self.netdir.hs_time_period(); let hs_dirs = self.netdir.hs_dirs_download( self.hs_blind_id, working_tp, &mut self.mocks.thread_rng(), )?; trace!( "HS desc fetch for {}, for period {}, using {} hsdirs", &self.hsid, working_tp, hs_dirs.len() ); let hs_dirs = hs_dirs .into_iter() .filter(|hsdir| { // Skip over any HsDirs that we are not allowed to requery right now let should_skip = recent_hsdirs.keys().any(|recent| { RelayIdForRequeryPeriod::for_lookup(hsdir).any(|id| id == *recent) }); !should_skip }) .collect::>(); if hs_dirs.is_empty() { warn!( "Tried to fetch HS desc for {}, for period {}, but all hsdirs are rate-limited", &self.hsid, working_tp, ); if stored_revision.is_none() { // We can't fetch a new descriptor, and we don't have a cached one. return Err(CE::NoUsableHsDirs); } else { // Return our cached descriptor return Ok(unwrap_valid_desc(data)); } } let params = self.netdir.params(); // We might consider launching requests to multiple HsDirs in parallel. // https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/1118#note_2894463 // But C Tor doesn't and our HS experts don't consider that important: // https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914436 // (Additionally, making multiple HSDir requests at once may make us // more vulnerable to traffic analysis.) let mut attempts = hs_dirs.iter().cycle().take(max_total_attempts); let mut errors = RetryError::in_attempt_to("retrieve hidden service descriptor"); let desc = loop { let relay = match attempts.next() { Some(relay) => relay, None => { return Err(if errors.is_empty() { CE::NoHsDirs } else { CE::DescriptorDownload(errors) }); } }; let hsdir_for_error: Sensitive = (*relay.id()).into(); let hsdir = RelayIdForRequeryPeriod::for_store(relay)?; // Ensure we wait at least hs_dir_requery_interval() until we try to // fecth from this HsDir again recent_hsdirs.insert(hsdir, now + self.config.retry.hs_dir_requery_interval()); match self.descriptor_fetch_attempt(relay, params).await { Ok(desc) => break desc, Err(error) => { if error.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!( &error, "Suspicious failure while downloading hsdesc for {} from relay {}", &self.hsid, relay.display_relay_ids(), ); } else { debug_report!( &error, "failed hsdir desc fetch for {} from {}/{}", &self.hsid, &relay.id(), &relay.rsa_id() ); } errors.push_timed( tor_error::Report(DescriptorError { hsdir: hsdir_for_error, error, }), self.runtime.now(), Some(self.runtime.wallclock()), ); } } }; // If our existing descriptor is newer than the one we have just fetched, // we should retain it. if let Some(stored_revision) = stored_revision { // It is safe to dangerously_assume_timely, // as descriptor_fetch_attempt has already checked the timeliness of the descriptor. let new_desc = desc.as_ref().dangerously_assume_timely(); // Revision counters are monotonically increasing within a given time period. // If our newly fetched descriptor has the same HsBlindId as our cached one, // it means they are both used for the same time period, // and so we should only update our cache if the new descriptor is more recent // (i.e. it has a higher revision counter). if stored_revision >= (new_desc.revision(), working_tp) { // Our cached descriptor is still timely, and has a higher revision counter // than the one we've just fetched, so we retain it. return Ok(unwrap_valid_desc(data)); } } // Store the bounded value in the cache for reuse, // but return a reference to the unwrapped `HsDesc`. // // The `HsDesc` must be owned by `data.desc`, // so first add it to `data.desc`, // and then dangerously_assume_timely to get a reference out again. // // It is safe to dangerously_assume_timely, // as descriptor_fetch_attempt has already checked the timeliness of the descriptor. let desc = HsDescForTp { time_period: working_tp, desc, }; let ret = data.insert(desc); Ok(ret.desc.as_ref().dangerously_assume_timely()) } /// Make one attempt to fetch the descriptor from a specific hsdir /// /// No timeout /// /// On success, returns the descriptor. /// /// While the returned descriptor is `TimeRangeBound`, its validity at the current time *has* /// been checked. #[instrument(level = "trace", skip_all)] async fn descriptor_fetch_attempt( &self, hsdir: &Relay<'_>, params: &NetParameters, ) -> Result, DescriptorErrorDetail> { let max_len: usize = self .netdir .params() .hsdir_max_desc_size .get() .try_into() .map_err(into_internal!("BoundedInt was not truly bounded!"))?; let request = { let mut r = tor_dirclient::request::HsDescDownloadRequest::new(self.hs_blind_id); r.set_max_len(max_len); r }; trace!( "hsdir for {}, trying {}/{}, request {:?} (http request {:?})", &self.hsid, &hsdir.id(), &hsdir.rsa_id(), &request, request.debug_request() ); let circuit = self .circpool .m_get_or_launch_dir(&self.netdir, OwnedCircTarget::from_circ_target(hsdir)) .await?; let n_hops = circuit.m_num_hops()?; let timeout_roundtrip = self.estimate_timeout(&[(1, TimeoutsAction::RoundTrip { length: n_hops })]); let source: Option = circuit .m_source_info() .map_err(into_internal!("Couldn't get SourceInfo for circuit"))?; let mut stream = self .runtime // NOTE: In fact this timeout is overkill: this operation should succeed immediately, // since we always send BEGINDIR messages optimistically (without waiting for a reply). // But since our code is complex, and since it could become possible for this to block // if the circuit is saturated or we implement proposal 367 or something, // we may as well have _some_ timeout here. .timeout(timeout_roundtrip, circuit.m_begin_dir_stream()) .await? .map_err(DescriptorErrorDetail::Circuit)?; let request_future = tor_dirclient::send_request(self.runtime, &request, &mut stream, source); let response = self .runtime .timeout(timeout_roundtrip, request_future) .await? .map_err(|dir_error| match dir_error { tor_dirclient::Error::RequestFailed(rfe) => DescriptorErrorDetail::from(rfe.error), tor_dirclient::Error::CircMgr(ce) => into_internal!( "tor-dirclient complains about circmgr going wrong but we gave it a stream" )(ce) .into(), other => into_internal!( "tor-dirclient gave unexpected error, tor-hsclient code needs updating" )(other) .into(), })?; let desc_text = response.into_output_string().map_err(|rfe| rfe.error)?; let hsc_desc_enc = self.secret_keys.keys.ks_hsc_desc_enc.as_ref(); let now = self.runtime.wallclock(); let desc = HsDesc::parse_decrypt_validate( &desc_text, &self.hs_blind_id, &self.subcredential, hsc_desc_enc, )?; // Check the validity time (relied on by descriptor_ensure) desc.check_valid_at(&now)?; // Validate cc_sendme_inc, if present, is within ±1 of params.cc_sendme_inc. ensure_descriptor_compatible_with_params(desc.dangerously_peek(), params)?; Ok(desc) } /// Given the descriptor, try to connect to service /// /// Does all necessary retries, timeouts, etc. async fn intro_rend_connect( &self, desc: &HsDesc, data: &mut DataIpts, ) -> Result { // Maximum number of rendezvous/introduction attempts we'll make let max_total_attempts = self .config .retry .hs_intro_rend_attempts() .try_into() // User specified a very large u32. We must be downcasting it to 16bit! // let's give them as many retries as we can manage. .unwrap_or(usize::MAX); // We can't reliably distinguish IPT failure from RPT failure, so we iterate over IPTs // (best first) and each time use a random RPT. // We limit the number of rendezvous establishment attempts, separately, since we don't // try to talk to the intro pt until we've established the rendezvous circuit. let mut rend_attempts = 0..max_total_attempts; // But, we put all the errors into the same bucket, since we might have a mixture. let mut errors = RetryError::in_attempt_to("make circuit to hidden service"); // Note that IntroPtIndex is *not* the index into this Vec. // It is the index into the original list of introduction points in the descriptor. let mut usable_intros: Vec = desc .intro_points() .iter() .enumerate() .map(|(intro_index, intro_desc)| { let intro_index = intro_index.into(); let intro_target = ipt_to_circtarget(intro_desc, &self.netdir) .map_err(|error| FAE::UnusableIntro { error, intro_index })?; // Lack of TAIT means this clone let intro_target = OwnedCircTarget::from_circ_target(&intro_target); Ok::<_, FailedAttemptError>(UsableIntroPt { intro_index, intro_desc, intro_target, sort_rand: self.mocks.thread_rng().random(), }) }) .filter_map(|entry| match entry { Ok(y) => Some(y), Err(e) => { errors.push_timed(e, self.runtime.now(), Some(self.runtime.wallclock())); None } }) .collect_vec(); // Delete experience information for now-unlisted intro points // Otherwise, as the IPTs change `Data` might grow without bound, // if we keep reconnecting to the same HS. data.retain(|k, _v| { usable_intros .iter() .any(|ipt| RelayIdForExperience::for_lookup(&ipt.intro_target).any(|id| &id == k)) }); // Join with existing state recording our experiences, // sort by descending goodness, and then randomly // (so clients without any experience don't all pile onto the same, first, IPT) usable_intros.sort_by_key(|ipt: &UsableIntroPt| { let experience = RelayIdForExperience::for_lookup(&ipt.intro_target).find_map(|id| data.get(&id)); IptSortKey { outcome: experience.into(), sort_rand: ipt.sort_rand, } }); self.mocks.test_got_ipts(&usable_intros); let mut intro_attempts = usable_intros.iter().cycle().take(max_total_attempts); // We retain a rendezvous we managed to set up in here. That way if we created it, and // then failed before we actually needed it, we can reuse it. // If we exit with an error, we will waste it - but because we isolate things we do // for different services, it wouldn't be reusable anyway. let mut saved_rendezvous = None; // If we are using proof-of-work DoS mitigation, this chooses an // algorithm and initial effort, and adjusts that effort when we retry. let mut pow_client = HsPowClient::new(&self.hs_blind_id, desc); // We might consider making multiple INTRODUCE attempts to different // IPTs in parallel, and somehow aggregating the errors and // experiences. // However our HS experts don't consider that important: // https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914438 // Parallelizing our HsCircPool circuit building would likely have // greater impact. (See #1149.) loop { // When did we start doing things that depended on the IPT? // // Used for recording our experience with the selected IPT let mut ipt_use_started = None::; // Error handling inner async block (analogous to an IEFE): // * Ok(Some()) means this attempt succeeded // * Ok(None) means all attempts exhausted // * Err(error) means this attempt failed // let outcome = async { // We establish a rendezvous point first. Although it appears from reading // this code that this means we serialise establishment of the rendezvous and // introduction circuits, this isn't actually the case. The circmgr maintains // a pool of circuits. What actually happens in the "standing start" case is // that we obtain a circuit for rendezvous from the circmgr's pool, expecting // one to be available immediately; the circmgr will then start to build a new // one to replenish its pool, and that happens in parallel with the work we do // here - but in arrears. If the circmgr pool is empty, then we must wait. // // Perhaps this should be parallelised here. But that's really what the pool // is for, since we expect building the rendezvous circuit and building the // introduction circuit to take about the same length of time. // // We *do* serialise the ESTABLISH_RENDEZVOUS exchange, with the // building of the introduction circuit. That could be improved, at the cost // of some additional complexity here. // // Our HS experts don't consider it important to increase the parallelism: // https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914444 // https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914445 if saved_rendezvous.is_none() { debug!("hs conn to {}: setting up rendezvous point", &self.hsid); // Establish a rendezvous circuit. let Some(_): Option = rend_attempts.next() else { return Ok(None); }; saved_rendezvous = Some(self.establish_rendezvous().await?); } let Some(ipt) = intro_attempts.next() else { return Ok(None); }; let intro_index = ipt.intro_index; let is_single_onion_service = desc.is_single_onion_service(); let proof_of_work = match pow_client.solve().await { Ok(solution) => solution, Err(e) => { debug!( "failing to compute proof-of-work, trying without. ({:?})", e ); None } }; // We record how long things take, starting from here, as // as a statistic we'll use for the IPT in future. // This is stored in a variable outside this async block, // so that the outcome handling can use it. ipt_use_started = Some(self.runtime.now()); // No `Option::get_or_try_insert_with`, or we'd avoid this expect() let rend_pt_for_error = rend_pt_identity_for_error( &saved_rendezvous .as_ref() .expect("just made Some") .rend_relay, ); debug!( "hs conn to {}: RPT {}", &self.hsid, rend_pt_for_error.as_inner() ); let (rendezvous, introduced) = self.exchange_introduce(desc, ipt, &mut saved_rendezvous, proof_of_work) .await // TODO: Maybe try, once, to extend-and-reuse the intro circuit. // // If the introduction fails, the introduction circuit is in principle // still usable. We believe that in this case, C Tor extends the intro // circuit by one hop to the next IPT to try. That saves on building a // whole new 3-hop intro circuit. However, our HS experts tell us that // if introduction fails at one IPT it is likely to fail at the others too, // so that optimisation might reduce our network impact and time to failure, // but isn't likely to improve our chances of success. // // However, it's not clear whether this approach risks contaminating // the 2nd attempt with some fault relating to the introduction point. // The 1st ipt might also gain more knowledge about which HS we're talking to. // // TODO SPEC: Discuss extend-and-reuse HS intro circuit after nack ?; #[allow(unused_variables)] // it's *supposed* to be unused let saved_rendezvous = (); // don't use `saved_rendezvous` any more, use rendezvous let rend_pt = rend_pt_identity_for_error(&rendezvous.rend_relay); let circ = self.complete_rendezvous(ipt, rendezvous, introduced, is_single_onion_service) .await?; debug!( "hs conn to {}: RPT {} IPT {}: success", &self.hsid, rend_pt.as_inner(), intro_index, ); Ok::<_, FAE>(Some((intro_index, circ))) } .await; // Store the experience `outcome` we had with IPT `intro_index`, in `data` #[allow(clippy::unused_unit)] // -> () is here for error handling clarity let mut store_experience = |intro_index, outcome| -> () { (|| { let ipt = usable_intros .iter() .find(|ipt| ipt.intro_index == intro_index) .ok_or_else(|| internal!("IPT not found by index"))?; let id = RelayIdForExperience::for_store(&ipt.intro_target)?; let started = ipt_use_started.ok_or_else(|| { internal!("trying to record IPT use but no IPT start time noted") })?; let duration = self .runtime .now() .checked_duration_since(started) .ok_or_else(|| internal!("clock overflow calculating IPT use duration"))?; data.insert(id, IptExperience { duration, outcome }); Ok::<_, Bug>(()) })() .unwrap_or_else(|e| warn_report!(e, "error recording HS IPT use experience")); }; match outcome { Ok(Some((intro_index, y))) => { // Record successful outcome in Data store_experience(intro_index, Ok(())); return Ok(y); } Ok(None) => return Err(CE::Failed(errors)), Err(error) => { debug_report!(&error, "hs conn to {}: attempt failed", &self.hsid); // Record error outcome in Data, if in fact we involved the IPT // at all. The IPT information is be retrieved from `error`, // since only some of the errors implicate the introduction point. if let Some(intro_index) = error.intro_index() { store_experience(intro_index, Err(error.retry_time())); } errors.push_timed(error, self.runtime.now(), Some(self.runtime.wallclock())); // If we are using proof-of-work DoS mitigation, try harder next time pow_client.increase_effort(); } } } } /// Make one attempt to establish a rendezvous circuit /// /// This doesn't really depend on anything, /// other than (obviously) the isolation implied by our circuit pool. /// In particular it doesn't depend on the introduction point. /// /// Applies timeouts as appropriate. #[instrument(level = "trace", skip_all)] async fn establish_rendezvous(&'c self) -> Result, FAE> { let (rend_tunnel, rend_relay) = self .circpool .m_get_or_launch_client_rend(&self.netdir) .await .map_err(|error| FAE::RendezvousCircuitObtain { error })?; let rend_pt = rend_pt_identity_for_error(&rend_relay); let rend_cookie: RendCookie = self.mocks.thread_rng().random(); let message = EstablishRendezvous::new(rend_cookie); let (rend_established_tx, rend_established_rx) = proto_oneshot::channel(); let (rend2_tx, rend2_rx) = proto_oneshot::channel(); /// Handler which expects `RENDEZVOUS_ESTABLISHED` and then /// `RENDEZVOUS2`. Returns each message via the corresponding `oneshot`. struct Handler { /// Sender for a RENDEZVOUS_ESTABLISHED message. rend_established_tx: proto_oneshot::Sender, /// Sender for a RENDEZVOUS2 message. rend2_tx: proto_oneshot::Sender, } impl MsgHandler for Handler { fn handle_msg( &mut self, msg: AnyRelayMsg, ) -> Result { // The first message we expect is a RENDEZVOUS_ESTABALISHED. if self.rend_established_tx.still_expected() { self.rend_established_tx .deliver_expected_message(msg, MetaCellDisposition::Consumed) } else { self.rend2_tx .deliver_expected_message(msg, MetaCellDisposition::ConversationFinished) } } } debug!( "hs conn to {}: RPT {}: sending ESTABLISH_RENDEZVOUS", &self.hsid, rend_pt.as_inner(), ); let failed_map_err = |error| FAE::RendezvousEstablish { error, rend_pt: rend_pt.clone(), }; let handler = Handler { rend_established_tx, rend2_tx, }; let num_hops = rend_tunnel .m_num_own_hops() .map_err(|error| FAE::RendezvousCircuitObtain { error })?; let timeout_roundtrip = self.estimate_timeout(&[(1, TimeoutsAction::RoundTrip { length: num_hops })]); // TODO(conflux) This error handling is horrible. Problem is that this Mock system requires // to send back a tor_circmgr::Error while our reply handler requires a tor_proto::Error. // And unifying both is hard here considering it needs to be converted to yet another Error // type "FAE" so we have to do these hoops and jumps. rend_tunnel .m_start_conversation_last_hop(Some(message.into()), handler) .await .map_err(|e| { let proto_error = match e { tor_circmgr::Error::Protocol { error, .. } => error, _ => tor_proto::Error::CircuitClosed, }; FAE::RendezvousEstablish { error: proto_error, rend_pt: rend_pt.clone(), } })?; // `start_conversation` returns as soon as the control message has been sent. // We need to obtain the RENDEZVOUS_ESTABLISHED message, which is "returned" via the oneshot. let _: RendezvousEstablished = self .runtime .timeout(timeout_roundtrip, rend_established_rx.recv(failed_map_err)) .await .map_err( |_timeout: tor_rtcompat::TimeoutError| FAE::RendezvousEstablishTimeout { rend_pt: rend_pt.clone(), }, )??; debug!( "hs conn to {}: RPT {}: got RENDEZVOUS_ESTABLISHED", &self.hsid, rend_pt.as_inner(), ); Ok(Rendezvous { rend_tunnel, rend_cookie, rend_relay, rend2_rx, marker: PhantomData, }) } /// Attempt (once) to send an INTRODUCE1 and wait for the INTRODUCE_ACK /// /// `take`s the input `rendezvous` (but only takes it if it gets that far) /// and, if successful, returns it. /// (This arranges that the rendezvous is "used up" precisely if /// we sent its secret somewhere.) /// /// Although this function handles the `Rendezvous`, /// nothing in it actually involves the rendezvous point. /// So if there's a failure, it's purely to do with the introduction point. /// /// Applies timeouts as appropriate. #[allow(clippy::type_complexity)] // TODO: Refactor #[instrument(level = "trace", skip_all)] async fn exchange_introduce( &'c self, hsdesc: &HsDesc, ipt: &UsableIntroPt<'_>, rendezvous: &mut Option>, proof_of_work: Option, ) -> Result<(Rendezvous<'c, R, M>, Introduced), FAE> { let intro_index = ipt.intro_index; debug!( "hs conn to {}: IPT {}: obtaining intro circuit", &self.hsid, intro_index, ); let intro_circ = self .circpool .m_get_or_launch_intro( &self.netdir, ipt.intro_target.clone(), // &OwnedCircTarget isn't CircTarget apparently ) .await .map_err(|error| FAE::IntroductionCircuitObtain { error, intro_index })?; let rendezvous = rendezvous.take().ok_or_else(|| internal!("no rend"))?; let rend_pt = rend_pt_identity_for_error(&rendezvous.rend_relay); debug!( "hs conn to {}: RPT {} IPT {}: making introduction", &self.hsid, rend_pt.as_inner(), intro_index, ); // Now we construct an introduce1 message and perform the first part of the // rendezvous handshake. // // This process is tricky because the header of the INTRODUCE1 message // -- which depends on the IntroPt configuration -- is authenticated as // part of the HsDesc handshake. // Construct the header, since we need it as input to our encryption. let intro_header = { let ipt_sid_key = ipt.intro_desc.ipt_sid_key(); let intro1 = Introduce1::new( AuthKeyType::ED25519_SHA3_256, ipt_sid_key.as_bytes().to_vec(), vec![], ); let mut header = vec![]; intro1 .encode_onto(&mut header) .map_err(into_internal!("couldn't encode intro1 header"))?; header }; let peer_caps = caps::PeerCaps::new(hsdesc); // Construct the introduce payload, which tells the onion service how to find // our rendezvous point. (We could do this earlier if we wanted.) let intro_payload = { let onion_key = intro_payload::OnionKey::NtorOnionKey(*rendezvous.rend_relay.ntor_onion_key()); let linkspecs = rendezvous .rend_relay .linkspecs() .map_err(into_internal!("Couldn't encode link specifiers"))?; #[allow(unused_mut)] // TODO: Remove once negotiate-extensions is always-on. let mut payload = IntroduceHandshakePayload::new( rendezvous.rend_cookie, onion_key, linkspecs, proof_of_work, ); peer_caps.add_extensions(&mut payload); let mut encoded = vec![]; payload .write_onto(&mut encoded) .map_err(into_internal!("Couldn't encode introduce1 payload"))?; encoded }; // Perform the cryptographic handshake with the onion service. let service_info = hs_ntor::HsNtorServiceInfo::new( ipt.intro_desc.svc_ntor_key().clone(), ipt.intro_desc.ipt_sid_key().clone(), self.subcredential, ); let handshake_state = hs_ntor::HsNtorClientState::new(&mut self.mocks.thread_rng(), service_info); let encrypted_body = handshake_state .client_send_intro(&intro_header, &intro_payload) .map_err(into_internal!("can't begin hs-ntor handshake"))?; // Build our actual INTRODUCE1 message. let intro1_real = Introduce1::new( AuthKeyType::ED25519_SHA3_256, ipt.intro_desc.ipt_sid_key().as_bytes().to_vec(), encrypted_body, ); /// Handler which expects just `INTRODUCE_ACK` struct Handler { /// Sender for `INTRODUCE_ACK` intro_ack_tx: proto_oneshot::Sender, } impl MsgHandler for Handler { fn handle_msg( &mut self, msg: AnyRelayMsg, ) -> Result { self.intro_ack_tx .deliver_expected_message(msg, MetaCellDisposition::ConversationFinished) } } let failed_map_err = |error| FAE::IntroductionExchange { error, intro_index }; let (intro_ack_tx, intro_ack_rx) = proto_oneshot::channel(); let handler = Handler { intro_ack_tx }; let num_hops = intro_circ .m_num_hops() .map_err(|error| FAE::IntroductionCircuitObtain { error, intro_index })?; // NOTE: Should we allow this to be longer in case the introduction point is grievously // overloaded? let timeout_roundtrip = self.estimate_timeout(&[(1, TimeoutsAction::RoundTrip { length: num_hops })]); debug!( "hs conn to {}: RPT {} IPT {}: making introduction - sending INTRODUCE1", &self.hsid, rend_pt.as_inner(), intro_index, ); // TODO(conflux) This error handling is horrible. Problem is that this Mock system requires // to send back a tor_circmgr::Error while our reply handler requires a tor_proto::Error. // And unifying both is hard here considering it needs to be converted to yet another Error // type "FAE" so we have to do these hoops and jumps. intro_circ .m_start_conversation_last_hop(Some(intro1_real.into()), handler) .await .map_err(|e| { let proto_error = match e { tor_circmgr::Error::Protocol { error, .. } => error, _ => tor_proto::Error::CircuitClosed, }; FAE::IntroductionExchange { error: proto_error, intro_index, } })?; // Status is checked by `.success()`, and we don't look at the extensions; // just discard the known-successful `IntroduceAck` let _: IntroduceAck = self .runtime .timeout(timeout_roundtrip, intro_ack_rx.recv(failed_map_err)) .await .map_err(|_timeout: TimeoutError| FAE::IntroductionTimeout { intro_index })?? .success() .map_err(|status| FAE::IntroductionFailed { status, intro_index, })?; debug!( "hs conn to {}: RPT {} IPT {}: making introduction - success", &self.hsid, rend_pt.as_inner(), intro_index, ); // Having received INTRODUCE_ACK. we can forget about this circuit // (and potentially tear it down). drop(intro_circ); Ok(( rendezvous, Introduced { handshake_state, marker: PhantomData, peer_caps, }, )) } /// Attempt (once) to connect a rendezvous circuit using the given intro pt. /// /// That is to say, we simply wait for a RENDEZVOUS2 message, /// and if we get one, we add a virtual hop. /// /// Timeouts here might be due to the IPT, RPT, service, /// or any of the intermediate relays. /// /// If, rather than a timeout, we actually encounter some kind of error, /// we'll return the appropriate `FailedAttemptError`. /// (Who is responsible may vary, so the `FailedAttemptError` variant will reflect that.) async fn complete_rendezvous( &'c self, ipt: &UsableIntroPt<'_>, rendezvous: Rendezvous<'c, R, M>, introduced: Introduced, is_single_onion_service: bool, ) -> Result { /// Largest number of hops that the onion service must build for _its_ /// circuits to our rendezvous points. /// /// This is 4 hops (assuming that it has full vanguards enabled) plus one for the /// renedezvous point itself. const MAX_PEER_REND_HOPS: usize = 5; /// Largest number of retries that we think the peer might make if its /// circuits are failing. const MAX_PEER_CIRC_RETRIES: u32 = 3; let rend_pt = rend_pt_identity_for_error(&rendezvous.rend_relay); let intro_index = ipt.intro_index; let failed_map_err = |error| FAE::RendezvousCompletionCircuitError { error, intro_index, rend_pt: rend_pt.clone(), }; debug!( "hs conn to {}: RPT {} IPT {}: awaiting rendezvous completion", &self.hsid, rend_pt.as_inner(), intro_index, ); let num_hops = rendezvous .rend_tunnel .m_num_own_hops() // This is not necessarily the best error, but it isn't totally wrong. // We can't wrap the tor_circuit error in anything else that makes sense. // See #2513. .map_err(|error| FAE::RendezvousCircuitObtain { error })?; // Maximum length of the circuit that the peer will build to the rendezvous point. let peer_rend_circ_len = if is_single_onion_service { 1 } else { MAX_PEER_REND_HOPS }; // The total number of hops from the peer to us. // // We subtract 1 because both circuits terminate at the rendezvous point. let total_circ_len = peer_rend_circ_len + num_hops - 1; // Limit on the duration of each attempt for activities involving both // RPT and IPT. let rpt_ipt_timeout = self.estimate_timeout(&[ // The API requires us to specify a number of circuit builds and round trips. // So what we tell the estimator is a rather imprecise description. // // What we are timing here is: // // INTRODUCE2 goes from IPT to HS. // This happens in parallel with our waiting for the INTRODUCE_ACK, // and we know that our own introduction circuit is always at least // as long as the peer's (even if they are using full vanguards), // so we don't need any additional delay here. // // HS builds to our RPT ( MAX_PEER_CIRC_RETRIES, TimeoutsAction::BuildCircuit { length: peer_rend_circ_len, }, ), // // RENDEZVOUS1 goes from HS to RPT. `peer_circ_len`, one-way. // RENDEZVOUS2 goes from RPT to us. `num_hops`, one-way. ( 1, TimeoutsAction::OneWay { length: total_circ_len, }, ), ]); let rend2_msg: Rendezvous2 = self .runtime .timeout(rpt_ipt_timeout, rendezvous.rend2_rx.recv(failed_map_err)) .await .map_err(|_: TimeoutError| FAE::RendezvousCompletionTimeout { intro_index, rend_pt: rend_pt.clone(), })??; debug!( "hs conn to {}: RPT {} IPT {}: received RENDEZVOUS2", &self.hsid, rend_pt.as_inner(), intro_index, ); // In theory would be great if we could have multiple introduction attempts in parallel // with similar x,X values but different IPTs. However, our HS experts don't // think increasing parallelism here is important: // https://gitlab.torproject.org/tpo/core/arti/-/issues/913#note_2914438 let handshake_state = introduced.handshake_state; // Try to complete the cryptographic handshake. let keygen = self.mocks .rendezvous_handshake(handshake_state, rend2_msg, intro_index, &rend_pt)?; let mut params = onion_circparams_from_netparams(self.netdir.params()) .map_err(into_internal!("Failed to build CircParameters"))?; if let Some(inc) = introduced.peer_caps.cc_sendme_inc() { params.ccontrol.override_sendme_inc(inc); } let protocols = introduced.peer_caps.shared_protos(); rendezvous .rend_tunnel .m_extend_virtual( handshake::RelayProtocol::HsV3, handshake::HandshakeRole::Initiator, keygen, params, &protocols, ) .await .map_err(into_internal!( "actually this is probably a 'circuit closed' error" // TODO HS ))?; debug!( "hs conn to {}: RPT {} IPT {}: HS circuit established", &self.hsid, rend_pt.as_inner(), intro_index, ); Ok(rendezvous.rend_tunnel) } /// Helper to estimate a timeout for a complicated operation /// /// `actions` is a list of `(count, action)`, where each entry /// represents doing `action`, `count` times sequentially. /// /// Combines the timeout estimates and returns an overall timeout. fn estimate_timeout(&self, actions: &[(u32, TimeoutsAction)]) -> Duration { // This algorithm is, perhaps, wrong. For uncorrelated variables, a particular // percentile estimate for a sum of random variables, is not calculated by adding the // percentile estimates of the individual variables. // // But the actual lengths of times of the operations aren't uncorrelated. // If they were *perfectly* correlated, then this addition would be correct. // It will do for now; it just might be rather longer than it ought to be. actions .iter() .map(|(count, action)| { self.circpool .m_estimate_timeout(action) .saturating_mul(*count) }) .fold(Duration::ZERO, Duration::saturating_add) } } /// Mocks used for testing `connect.rs` /// /// This is different to `MockableConnectorData`, /// which is used to *replace* this file, when testing `state.rs`. /// /// `MocksForConnect` provides mock facilities for *testing* this file. // // TODO this should probably live somewhere else, maybe tor-circmgr even? // TODO this really ought to be made by macros or something trait MocksForConnect: Clone { /// HS circuit pool type HsCircPool: MockableCircPool; /// A random number generator type Rng: rand::Rng + rand::CryptoRng; /// Key generator used for generating the keys for the virtual hop. type KeyGenerator: tor_proto::client::circuit::handshake::KeyGenerator + Send; /// Tell tests we got this descriptor text fn test_got_desc(&self, _: &HsDesc) {} /// Tell tests we got this data tunnel. fn test_got_tunnel(&self, _: &DataTunnel!(R, Self)) {} /// Tell tests we have obtained and sorted the intros like this fn test_got_ipts(&self, _: &[UsableIntroPt]) {} /// Return a random number generator fn thread_rng(&self) -> Self::Rng; /// Complete the rendezvous handshake, returning the resulting keygen fn rendezvous_handshake( &self, handshake_state: hs_ntor::HsNtorClientState, rend2_msg: Rendezvous2, intro_index: IntroPtIndex, rend_pt: &RendPtIdentityForError, ) -> Result; } /// Mock for `HsCircPool` /// /// Methods start with `m_` to avoid the following problem: /// `ClientCirc::start_conversation` (say) means /// to use the inherent method if one exists, /// but will use a trait method if there isn't an inherent method. /// /// So if the inherent method is renamed, the call in the impl here /// turns into an always-recursive call. /// This is not detected by the compiler due to the situation being /// complicated by futures, `#[async_trait]` etc. /// #[async_trait] trait MockableCircPool { /// Directory tunnel. type DirTunnel: MockableClientDir; /// Data tunnel. type DataTunnel: MockableClientData; /// Intro tunnel. type IntroTunnel: MockableClientIntro; async fn m_get_or_launch_dir( &self, netdir: &NetDir, target: impl CircTarget + Send + Sync + 'async_trait, ) -> tor_circmgr::Result; async fn m_get_or_launch_intro( &self, netdir: &NetDir, target: impl CircTarget + Send + Sync + 'async_trait, ) -> tor_circmgr::Result; /// Client circuit async fn m_get_or_launch_client_rend<'a>( &self, netdir: &'a NetDir, ) -> tor_circmgr::Result<(Self::DataTunnel, Relay<'a>)>; /// Estimate timeout fn m_estimate_timeout(&self, action: &TimeoutsAction) -> Duration; } /// Mock for onion service client directory tunnel. #[async_trait] trait MockableClientDir: Debug { /// Client circuit type DirStream: AsyncRead + AsyncWrite + Send + Unpin; async fn m_begin_dir_stream(&self) -> tor_circmgr::Result; /// Get a tor_dirclient::SourceInfo for this circuit, if possible. fn m_source_info(&self) -> tor_proto::Result>; /// Return the length of this circuit. fn m_num_hops(&self) -> tor_circmgr::Result; } /// Mock for onion service client data tunnel. #[async_trait] trait MockableClientData: Debug { /// Conversation type Conversation<'r> where Self: 'r; /// Converse async fn m_start_conversation_last_hop( &self, msg: Option, reply_handler: impl MsgHandler + Send + 'static, ) -> tor_circmgr::Result>; /// Add a virtual hop to the circuit. async fn m_extend_virtual( &self, protocol: handshake::RelayProtocol, role: handshake::HandshakeRole, handshake: impl handshake::KeyGenerator + Send, params: CircParameters, capabilities: &tor_protover::Protocols, ) -> tor_circmgr::Result<()>; /// Return the number of our own hops in this circuit. /// /// This does not count any hops for the service's rendezvous circuit. /// It does count our virtual hop, if we have one. /// (That isn't a problem, since we only use this method to calculate /// timeouts, and we only calculate timeouts _before_ we establish /// the virtual hop.) fn m_num_own_hops(&self) -> tor_circmgr::Result; } /// Mock for onion service client introduction tunnel. #[async_trait] trait MockableClientIntro: Debug { /// Conversation type Conversation<'r> where Self: 'r; /// Converse async fn m_start_conversation_last_hop( &self, msg: Option, reply_handler: impl MsgHandler + Send + 'static, ) -> tor_circmgr::Result>; /// Return the number of hops in this circuit. fn m_num_hops(&self) -> tor_circmgr::Result; } impl MocksForConnect for () { type HsCircPool = HsCircPool; type Rng = rand::rngs::ThreadRng; type KeyGenerator = HsNtorHkdfKeyGenerator; fn thread_rng(&self) -> Self::Rng { rand::rng() } fn rendezvous_handshake( &self, handshake_state: hs_ntor::HsNtorClientState, rend2_msg: Rendezvous2, intro_index: IntroPtIndex, rend_pt: &RendPtIdentityForError, ) -> Result { // Try to complete the cryptographic handshake. handshake_state .client_receive_rend(rend2_msg.handshake_info()) // If this goes wrong. either the onion service has mangled the crypto, // or the rendezvous point has misbehaved (that that is possible is a protocol bug), // or we have used the wrong handshake_state (let's assume that's not true). // // If this happens we'll go and try another RPT. .map_err(|error| FAE::RendezvousCompletionHandshake { error, intro_index, rend_pt: rend_pt.clone(), }) } } #[async_trait] impl MockableCircPool for HsCircPool { type DirTunnel = ClientOnionServiceDirTunnel; type DataTunnel = ClientOnionServiceDataTunnel; type IntroTunnel = ClientOnionServiceIntroTunnel; #[instrument(level = "trace", skip_all)] async fn m_get_or_launch_dir( &self, netdir: &NetDir, target: impl CircTarget + Send + Sync + 'async_trait, ) -> tor_circmgr::Result { Ok(HsCircPool::get_or_launch_client_dir(self, netdir, target).await?) } #[instrument(level = "trace", skip_all)] async fn m_get_or_launch_intro( &self, netdir: &NetDir, target: impl CircTarget + Send + Sync + 'async_trait, ) -> tor_circmgr::Result { Ok(HsCircPool::get_or_launch_client_intro(self, netdir, target).await?) } #[instrument(level = "trace", skip_all)] async fn m_get_or_launch_client_rend<'a>( &self, netdir: &'a NetDir, ) -> tor_circmgr::Result<(Self::DataTunnel, Relay<'a>)> { HsCircPool::get_or_launch_client_rend(self, netdir).await } fn m_estimate_timeout(&self, action: &TimeoutsAction) -> Duration { HsCircPool::estimate_timeout(self, action) } } #[async_trait] impl MockableClientDir for ClientOnionServiceDirTunnel { /// Client circuit type DirStream = tor_proto::client::stream::DataStream; async fn m_begin_dir_stream(&self) -> tor_circmgr::Result { Self::begin_dir_stream(self).await } /// Get a tor_dirclient::SourceInfo for this circuit, if possible. fn m_source_info(&self) -> tor_proto::Result> { SourceInfo::from_tunnel(self) } fn m_num_hops(&self) -> tor_circmgr::Result { self.n_hops() } } #[async_trait] impl MockableClientData for ClientOnionServiceDataTunnel { type Conversation<'r> = tor_proto::Conversation<'r>; async fn m_start_conversation_last_hop( &self, msg: Option, reply_handler: impl MsgHandler + Send + 'static, ) -> tor_circmgr::Result> { Self::start_conversation(self, msg, reply_handler, TargetHop::LastHop).await } async fn m_extend_virtual( &self, protocol: handshake::RelayProtocol, role: handshake::HandshakeRole, handshake: impl handshake::KeyGenerator + Send, params: CircParameters, capabilities: &tor_protover::Protocols, ) -> tor_circmgr::Result<()> { Self::extend_virtual(self, protocol, role, handshake, params, capabilities).await } fn m_num_own_hops(&self) -> tor_circmgr::Result { self.n_hops() } } #[async_trait] impl MockableClientIntro for ClientOnionServiceIntroTunnel { type Conversation<'r> = tor_proto::Conversation<'r>; async fn m_start_conversation_last_hop( &self, msg: Option, reply_handler: impl MsgHandler + Send + 'static, ) -> tor_circmgr::Result> { Self::start_conversation(self, msg, reply_handler, TargetHop::LastHop).await } fn m_num_hops(&self) -> tor_circmgr::Result { self.n_hops() } } #[async_trait] impl MockableConnectorData for Data { type DataTunnel = ClientOnionServiceDataTunnel; type MockGlobalState = (); async fn connect( connector: &HsClientConnector, netdir: Arc, config: Arc, hsid: HsId, data: &mut Self, secret_keys: HsClientSecretKeys, ) -> Result { connect(connector, netdir, config, hsid, data, secret_keys).await } fn tunnel_is_ok(tunnel: &Self::DataTunnel) -> bool { !tunnel.is_closed() } } /// Return an error if `desc` declares a set of parameters /// that is too far away from the consensus. fn ensure_descriptor_compatible_with_params( desc: &HsDesc, params: &NetParameters, ) -> Result<(), DescriptorErrorDetail> { if let Some((_, desc_inc)) = desc.flow_control() { let difference = u8::from(desc_inc).abs_diff(params.cc_sendme_inc.into()); if difference > 1 { return Err(DescriptorErrorDetail::ParameterMismatch( "cc_sendme_inc too far from consensus".into(), )); } } Ok(()) } #[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 //! #![allow(dead_code, unused_variables)] // TODO HS TESTS delete, after tests are completed use super::*; use crate::*; use itertools::chain; use std::iter; use tokio_crate as tokio; use tor_async_utils::JoinReadWrite; use tor_basic_utils::test_rng::{TestingRng, testing_rng}; use tor_hscrypto::pk::{HsClientDescEncKey, HsClientDescEncKeypair}; use tor_llcrypto::pk::curve25519; use tor_netdoc::doc::{hsdesc::test_data, netstatus::Lifetime}; use tor_rtcompat::RuntimeSubstExt as _; use tor_rtcompat::tokio::TokioNativeTlsRuntime; use tor_rtmock::simple_time::SimpleMockTimeProvider; use tracing_test::traced_test; #[derive(derive_more::Debug, Default)] struct MocksGlobal { hsdirs_asked: Vec, got_desc: Option, #[debug(skip)] rendezvous: Option>, intro_acks: Vec<(IntroduceAck, MetaCellDisposition)>, } #[derive(Clone, Debug)] struct Mocks { mglobal: Arc>, id: I, } struct MockKeyGenerator; impl handshake::KeyGenerator for MockKeyGenerator { fn expand(self, _keylen: usize) -> tor_proto::Result { todo!() } } impl MocksForConnect for Mocks<()> { type HsCircPool = Mocks<()>; type Rng = TestingRng; type KeyGenerator = MockKeyGenerator; fn test_got_desc(&self, desc: &HsDesc) { self.mglobal.lock().unwrap().got_desc = Some(desc.clone()); } fn test_got_ipts(&self, desc: &[UsableIntroPt]) {} fn thread_rng(&self) -> Self::Rng { testing_rng() } fn rendezvous_handshake( &self, _handshake_state: hs_ntor::HsNtorClientState, _rend2_msg: Rendezvous2, _intro_index: IntroPtIndex, _rend_pt: &RendPtIdentityForError, ) -> Result { Ok(MockKeyGenerator) } } #[async_trait] impl MockableCircPool for Mocks<()> { type DataTunnel = Mocks<()>; type DirTunnel = Mocks<()>; type IntroTunnel = Mocks<()>; async fn m_get_or_launch_dir( &self, _netdir: &NetDir, target: impl CircTarget + Send + Sync + 'async_trait, ) -> tor_circmgr::Result { let target = OwnedCircTarget::from_circ_target(&target); self.mglobal.lock().unwrap().hsdirs_asked.push(target); Ok(self.clone()) } async fn m_get_or_launch_intro( &self, _netdir: &NetDir, target: impl CircTarget + Send + Sync + 'async_trait, ) -> tor_circmgr::Result { Ok(self.clone()) } /// Client circuit async fn m_get_or_launch_client_rend<'a>( &self, netdir: &'a NetDir, ) -> tor_circmgr::Result<(Self::DataTunnel, Relay<'a>)> { // Pick one of the relays we know to be in the test net as our RPT let rpt = netdir.by_id(&Ed25519Identity::from([12; 32])).unwrap(); Ok((self.clone(), rpt)) } fn m_estimate_timeout(&self, action: &TimeoutsAction) -> Duration { Duration::from_secs(10) } } #[async_trait] impl MockableClientDir for Mocks<()> { type DirStream = JoinReadWrite>, futures::io::Sink>; async fn m_begin_dir_stream(&self) -> tor_circmgr::Result { let response = format!( r#"HTTP/1.1 200 OK {}"#, test_data::TEST_DATA_2 ) .into_bytes() .into_boxed_slice(); Ok(JoinReadWrite::new( futures::io::Cursor::new(response), futures::io::sink(), )) } fn m_source_info(&self) -> tor_proto::Result> { Ok(None) } fn m_num_hops(&self) -> tor_circmgr::Result { Ok(4) } } #[async_trait] impl MockableClientData for Mocks<()> { type Conversation<'r> = &'r (); async fn m_start_conversation_last_hop( &self, msg: Option, mut reply_handler: impl MsgHandler + Send + 'static, ) -> tor_circmgr::Result> { match msg { Some(AnyRelayMsg::EstablishRendezvous(_)) => { let reply = RendezvousEstablished::default(); let disp = reply_handler.handle_msg(reply.into()).unwrap(); assert_eq!(disp, MetaCellDisposition::Consumed); // Save this, because we'll need to use it later, // when handling the INTRODUCE1 let mut global = self.mglobal.lock().unwrap(); global.rendezvous = Some(Box::new(reply_handler)); } _ => panic!("unexpected msg {msg:?}"), } Ok(&()) } async fn m_extend_virtual( &self, protocol: handshake::RelayProtocol, role: handshake::HandshakeRole, handshake: impl handshake::KeyGenerator + Send, params: CircParameters, capabilities: &tor_protover::Protocols, ) -> tor_circmgr::Result<()> { Ok(()) } fn m_num_own_hops(&self) -> tor_circmgr::Result { Ok(4) } } #[async_trait] impl MockableClientIntro for Mocks<()> { type Conversation<'r> = &'r (); async fn m_start_conversation_last_hop( &self, msg: Option, mut reply_handler: impl MsgHandler + Send + 'static, ) -> tor_circmgr::Result> { match msg { Some(AnyRelayMsg::Introduce1(introduce1)) => { let mut global = self.mglobal.lock().unwrap(); let (reply, expected_disp) = global.intro_acks.remove(0); let disp = reply_handler.handle_msg(reply.into()).unwrap(); assert_eq!(disp, expected_disp); // Mock the service's response let rendezvous = global .rendezvous .as_mut() .expect("got INTRODUCE1 before ESTABLISH_RENDEZVOUS?!"); let reply = Rendezvous2::new(b"dummy handshake info, ignored"); let disp = rendezvous.handle_msg(reply.into()).unwrap(); assert_eq!(disp, MetaCellDisposition::ConversationFinished); } _ => panic!("unexpected msg {msg:?}"), } Ok(&()) } fn m_num_hops(&self) -> tor_circmgr::Result { Ok(4) } } fn ks_hsc_desc_enc() -> HsClientDescEncKeypair { let pk: HsClientDescEncKey = curve25519::PublicKey::from(test_data::TEST_PUBKEY_2).into(); let sk = curve25519::StaticSecret::from(test_data::TEST_SECKEY_2).into(); HsClientDescEncKeypair::new(pk, sk) } fn expected_hsdesc(hsid: HsId, netdir: &NetDir, now: SystemTime) -> HsDesc { let time_period = netdir.hs_time_period(); let (hs_blind_id_key, subcredential) = HsIdKey::try_from(hsid) .unwrap() .compute_blinded_key(time_period) .unwrap(); let hs_blind_id = hs_blind_id_key.id(); HsDesc::parse_decrypt_validate( test_data::TEST_DATA_2, &hs_blind_id, &subcredential, Some(&ks_hsc_desc_enc()), ) .unwrap() .if_valid_at(&now) .unwrap() } fn build_test_netdir() -> Arc { let valid_after = humantime::parse_rfc3339("2023-02-09T12:00:00Z").unwrap(); let fresh_until = valid_after + humantime::parse_duration("1 hours").unwrap(); let valid_until = valid_after + humantime::parse_duration("24 hours").unwrap(); let lifetime = Lifetime::new(valid_after, fresh_until, valid_until).unwrap(); let netdir = tor_netdir::testnet::construct_custom_netdir_with_params( tor_netdir::testnet::simple_net_func, iter::empty::<(&str, _)>(), Some(lifetime), ) .expect("failed to build default testing netdir"); Arc::new(netdir.unwrap_if_sufficient().unwrap()) } #[traced_test] #[tokio::test] async fn test_connect() { use MetaCellDisposition::*; let netdir = build_test_netdir(); let runtime = TokioNativeTlsRuntime::current().unwrap(); let now = humantime::parse_rfc3339("2023-02-09T12:00:00Z").unwrap(); let mock_sp = SimpleMockTimeProvider::from_wallclock(now); let runtime = runtime .with_sleep_provider(mock_sp.clone()) .with_coarse_time_provider(mock_sp.clone()); let success = ( IntroduceAck::new(IntroduceAckStatus::SUCCESS), ConversationFinished, ); let nack = ( IntroduceAck::new(IntroduceAckStatus::NOT_RECOGNIZED), ConversationFinished, ); // The number of times to make Context:connect() fail due to intro NACK // // Set to 5 in order to trigger a rate-limit for all 6 HsDirs: // // there are 6 HsDirs in total, one of which is "used up" by the // first (successful) connect() attempt below. const INTRO_FAIL_COUNT: usize = 5; /// The number of times we expect the client to retry the /// introduction per connect() call /// (it will essentially try two rounds of `intro_rend_connect()`, /// once with the cached descriptor, and once with the potentially /// new descriptor). const IPT_RETRY_COUNT: usize = 12; // The first introduction will succeed let intro_acks = chain!( [&success], // But the next INTRO_FAIL_COUNT connect() will fail // (+1 because we want to fail *again*, in order to find // that there's now a limit on all our HsDirs) [&nack; IPT_RETRY_COUNT * (INTRO_FAIL_COUNT + 1)], // One more round of failures, to trigger a refecth after the rate-limit is lifted [&nack; IPT_RETRY_COUNT - 1], // After refetching the descriptor, the client will retry the introduction, // and succeed. [&success], ) .cloned() .collect(); let mglobal = Arc::new(Mutex::new(MocksGlobal { intro_acks, ..Default::default() })); let mocks = Mocks { mglobal, id: () }; // From C Tor src/test/test_hs_common.c test_build_address let hsid = test_data::TEST_HSID_2.into(); let mut data = Data::default(); let mut expected_hsdirs_asked = 1; let mut secret_keys_builder = HsClientSecretKeysBuilder::default(); secret_keys_builder.ks_hsc_desc_enc(ks_hsc_desc_enc()); let secret_keys = secret_keys_builder.build().unwrap(); let ctx = Context::new( &runtime, &mocks, Arc::clone(&netdir), Default::default(), hsid, secret_keys, mocks.clone(), ) .unwrap(); let _got = ctx.connect(&mut data).await.unwrap(); // Our mock IPT hasn't sent any NACKs yet assert!(!logs_contain("NACKed, refetching descriptor and retrying")); let hsdesc = expected_hsdesc(hsid, &netdir, now); { let mglobal = mocks.mglobal.lock().unwrap(); assert_eq!(mglobal.hsdirs_asked.len(), expected_hsdirs_asked); // TODO hs: here and in other places, consider implementing PartialEq instead, or creating // an assert_dbg_eq macro (which would be part of a test_helpers crate or something) assert_eq!( format!("{:?}", mglobal.got_desc), format!("{:?}", Some(hsdesc.clone())) ); } // Check how long the descriptor is valid for let (start_time, end_time) = data.desc.as_ref().unwrap().desc.bounds_start_end(); assert_eq!(start_time, None); let desc_valid_until = humantime::parse_rfc3339("2023-02-11T20:00:00Z").unwrap(); assert_eq!(end_time, Some(desc_valid_until)); // These attempts will all fail due to intro NACK, // and trigger a rate-limit for all 6 HsDirs for i in 1..=INTRO_FAIL_COUNT + 1 { let err = ctx.connect(&mut data).await.unwrap_err(); let is_intro_nack = |e| matches!(e, FAE::IntroductionFailed { status, .. }); // All attempts failed because of our repeated intro NACKs assert!(matches!(err, CE::Failed(e) if e.clone().into_iter().all(is_intro_nack))); { assert!(logs_contain("NACKed, refetching descriptor and retrying")); let mglobal = mocks.mglobal.lock().unwrap(); // Because all intro attempts failed with NACK (NOT_RECOGNIZED), // the client must've tried to refetch the descriptor if i <= INTRO_FAIL_COUNT { // No rate limiting yet, so the client must've tried to fetch a new // descriptor, before failing again. expected_hsdirs_asked += 1; assert!(!logs_contain("but all hsdirs are rate-limited")); assert_eq!(mglobal.hsdirs_asked.len(), expected_hsdirs_asked); } else { // The final failure won't lead to an HsDir fetch // because all HsDirs will be rate-limited at that point assert!(logs_contain("but all hsdirs are rate-limited")); assert_eq!(mglobal.hsdirs_asked.len(), expected_hsdirs_asked); } // Same descriptor each time // TODO hs: here and in other places, consider implementing PartialEq instead, or creating // an assert_dbg_eq macro (which would be part of a test_helpers crate or something) assert_eq!( format!("{:?}", mglobal.got_desc), format!("{:?}", Some(hsdesc.clone())) ); } let (start_time, end_time) = data.desc.as_ref().unwrap().desc.bounds_start_end(); assert_eq!(start_time, None); let desc_valid_until = humantime::parse_rfc3339("2023-02-11T20:00:00Z").unwrap(); assert_eq!(end_time, Some(desc_valid_until)); } // By default, the HsDir fetches are rate-limited for 15min mock_sp.advance(Duration::from_secs(15 * 60)); // Finally, we succeed. let _got = ctx.connect(&mut data).await.unwrap(); // And it turns out we did, in fact refetch the descriptor // Finally, we try again, but find that all HsDirs are now rate-limited! // So now we advance the time to lift the rate limit, and hope that // // TODO HS TESTS: we could extend our mock infrastructure // to support returning a different hsdesc this time, // with various revision counters, to check that the client is indeed // keeping the newest one. { assert!(logs_contain("NACKed, refetching descriptor and retrying")); let mglobal = mocks.mglobal.lock().unwrap(); // Because all intro attempts failed with NACK (NOT_RECOGNIZED), // the client must've tried to refetch the descriptor expected_hsdirs_asked += 1; assert_eq!(mglobal.hsdirs_asked.len(), expected_hsdirs_asked); } // TODO HS TESTS: check the circuit in got is the one we gave out // TODO HS TESTS: continue with this } // TODO HS TESTS: Test IPT state management and expiry: // - obtain a test descriptor with only a broken ipt // (broken in the sense that intro can be attempted, but will fail somehow) // - try to make a connection and expect it to fail // - assert that the ipt data isn't empty // - cause the descriptor to expire (advance clock) // - start using a mocked RNG if we weren't already and pin its seed here // - make a new descriptor with two IPTs: the broken one from earlier, and a new one // - make a new connection // - use test_got_ipts to check that the random numbers // would sort the bad intro first, *and* that the good one is appears first // - assert that connection succeeded // - cause the circuit and descriptor to expire (advance clock) // - go back to the previous descriptor contents, but with a new validity period // - try to make a connection // - use test_got_ipts to check that only the broken ipt is present // TODO HS TESTS: test retries (of every retry loop we have here) // TODO HS TESTS: test error paths }