Merge branch 'dirmgr-purification-2' into 'main' (0597c31a) · Commits · Steven Murdoch / Arti

Cargo.lock

+1 −0

Original line number	Diff line number	Diff line
		@@ -3517,6 +3517,7 @@ dependencies = [
		"tor-netdir",
		"tor-netdoc",
		"tor-rtcompat",
		"tor-rtmock",
		"tracing",
		]

crates/tor-dirmgr/Cargo.toml

+1 −0

Original line number	Diff line number	Diff line
		@@ -69,4 +69,5 @@ futures-await-test = "0.3.0"
		hex-literal = "0.3"
		tempfile = "3"
		tor-rtcompat = { path = "../tor-rtcompat", version = "0.3.0", features = ["tokio", "native-tls"] }
		tor-rtmock = { path = "../tor-rtmock", version = "0.3.0" }
		float_eq = "0.7"

crates/tor-dirmgr/src/bootstrap.rs

+214 −54

Original line number	Diff line number	Diff line
		//! Functions to download or load directory objects, using the
		//! state machines in the `states` module.

		use std::ops::{Deref, DerefMut};
		use std::{
		collections::HashMap,
		sync::{Arc, Weak},
		time::{Duration, SystemTime},
		};

		use crate::state::DirState;
		use crate::{
		docid::{self, ClientRequest},
		state::WriteNetDir,
		upgrade_weak_ref, DirMgr, DirState, DocId, DocumentText, Error, Readiness, Result,
		upgrade_weak_ref, DirMgr, DocId, DocQuery, DocumentText, Error, Readiness, Result,
		};

		use futures::channel::oneshot;
		@@ -21,77 +22,217 @@ use tor_dirclient::DirResponse;
		use tor_rtcompat::{Runtime, SleepProviderExt};
		use tracing::{debug, info, trace, warn};

		use crate::storage::Store;
		#[cfg(test)]
		use once_cell::sync::Lazy;
		#[cfg(test)]
		use std::sync::Mutex;
		use tor_circmgr::{CircMgr, DirInfo};
		use tor_netdir::NetDir;
		use tor_netdoc::doc::netstatus::ConsensusFlavor;

		/// If there were errors from a peer in `outcome`, record those errors by
		/// marking the circuit (if any) as needing retirement, and noting the peer
		/// (if any) as having failed.
		fn note_request_outcome<R: Runtime>(
		circmgr: &CircMgr<R>,
		outcome: &tor_dirclient::Result<tor_dirclient::DirResponse>,
		) {
		use tor_dirclient::Error::RequestFailed;
		// Extract an error and a source from this outcome, if there is one.
		//
		// This is complicated because DirResponse can encapsulate the notion of
		// a response that failed part way through a download: in the case, it
		// has some data, and also an error.
		let (err, source) = match outcome {
		Ok(req) => {
		if let (Some(e), Some(source)) = (req.error(), req.source()) {
		(
		RequestFailed {
		error: e.clone(),
		source: Some(source.clone()),
		},
		source,
		)
		} else {
		return;
		}
		}
		Err(RequestFailed {
		source: Some(source),
		..
		}) => (
		// TODO: Use an @ binding in the pattern once we are on MSRV >=
		// 1.56.
		outcome.as_ref().unwrap_err().clone(),
		source,
		),
		_ => return,
		};

		note_cache_error(circmgr, source, &err.into());
		}

		/// Record that a problem has occurred because of a failure in an answer from `source`.
		fn note_cache_error<R: Runtime>(
		circmgr: &CircMgr<R>,
		source: &tor_dirclient::SourceInfo,
		problem: &Error,
		) {
		use tor_circmgr::ExternalActivity;

		if !problem.indicates_cache_failure() {
		return;
		}

		info!("Marking {:?} as failed: {}", source, problem);
		circmgr.note_external_failure(source.cache_id(), ExternalActivity::DirCache);
		circmgr.retire_circ(source.unique_circ_id());
		}

		/// Try to read a set of documents from `dirmgr` by ID.
		fn load_all<R: Runtime>(
		dirmgr: &DirMgr<R>,
		missing: Vec<DocId>,
		/// Record that `source` has successfully given us some directory info.
		fn note_cache_success<R: Runtime>(circmgr: &CircMgr<R>, source: &tor_dirclient::SourceInfo) {
		use tor_circmgr::ExternalActivity;

		trace!("Marking {:?} as successful", source);
		circmgr.note_external_success(source.cache_id(), ExternalActivity::DirCache);
		}

		/// Load a set of documents from a `Store`, returning all documents found in the store.
		/// Note that this may be less than the number of documents in `missing`.
		fn load_documents_from_store(
		missing: &[DocId],
		store: &dyn Store,
		) -> Result<HashMap<DocId, DocumentText>> {
		let mut loaded = HashMap::new();
		for query in docid::partition_by_type(missing.into_iter()).values() {
		dirmgr.load_documents_into(query, &mut loaded)?;
		for query in docid::partition_by_type(missing.iter().copied()).values() {
		query.load_from_store_into(&mut loaded, store)?;
		}
		Ok(loaded)
		}

		/// Testing helper: if this is Some, then we return it in place of any
		/// response to fetch_single.
		///
		/// Note that only one test uses this: otherwise there would be a race
		/// condition. :p
		#[cfg(test)]
		static CANNED_RESPONSE: Lazy<Mutex<Option<String>>> = Lazy::new(\|\| Mutex::new(None));
		/// Construct an appropriate ClientRequest to download a consensus
		/// of the given flavor.
		pub(crate) fn make_consensus_request(
		now: SystemTime,
		flavor: ConsensusFlavor,
		store: &dyn Store,
		) -> Result<ClientRequest> {
		let mut request = tor_dirclient::request::ConsensusRequest::new(flavor);

		let default_cutoff = crate::default_consensus_cutoff(now)?;

		match store.latest_consensus_meta(flavor) {
		Ok(Some(meta)) => {
		let valid_after = meta.lifetime().valid_after();
		request.set_last_consensus_date(std::cmp::max(valid_after, default_cutoff));
		request.push_old_consensus_digest(*meta.sha3_256_of_signed());
		}
		latest => {
		if let Err(e) = latest {
		warn!("Error loading directory metadata: {}", e);
		}
		// If we don't have a consensus, then request one that's
		// "reasonably new". That way, our clock is set far in the
		// future, we won't download stuff we can't use.
		request.set_last_consensus_date(default_cutoff);
		}
		}

		Ok(ClientRequest::Consensus(request))
		}

		/// Construct a set of `ClientRequest`s in order to fetch the documents in `docs`.
		pub(crate) fn make_requests_for_documents<R: Runtime>(
		rt: &R,
		docs: &[DocId],
		store: &dyn Store,
		) -> Result<Vec<ClientRequest>> {
		let mut res = Vec::new();
		for q in docid::partition_by_type(docs.iter().copied())
		.into_iter()
		.flat_map(\|(_, x)\| x.split_for_download().into_iter())
		{
		match q {
		DocQuery::LatestConsensus { flavor, .. } => {
		res.push(make_consensus_request(rt.wallclock(), flavor, store)?);
		}
		DocQuery::AuthCert(ids) => {
		res.push(ClientRequest::AuthCert(ids.into_iter().collect()));
		}
		DocQuery::Microdesc(ids) => {
		res.push(ClientRequest::Microdescs(ids.into_iter().collect()));
		}
		#[cfg(feature = "routerdesc")]
		DocQuery::RouterDesc(ids) => {
		res.push(ClientRequest::RouterDescs(ids.into_iter().collect()));
		}
		}
		}
		Ok(res)
		}

		/// Launch a single client request and get an associated response.
		async fn fetch_single<R: Runtime>(
		dirmgr: Arc<DirMgr<R>>,
		rt: &R,
		request: ClientRequest,
		current_netdir: Option<&NetDir>,
		circmgr: Arc<CircMgr<R>>,
		) -> Result<(ClientRequest, DirResponse)> {
		#[cfg(test)]
		{
		let m = CANNED_RESPONSE.lock().expect("Poisoned mutex");
		if let Some(s) = m.as_ref() {
		return Ok((request, DirResponse::from_body(s)));
		}
		}
		let circmgr = dirmgr.circmgr()?;
		let cur_netdir = dirmgr.opt_netdir();
		let dirinfo = match cur_netdir {
		Some(ref netdir) => netdir.as_ref().into(),
		let dirinfo: DirInfo = match current_netdir {
		Some(netdir) => netdir.into(),
		None => tor_circmgr::DirInfo::Nothing,
		};
		let outcome =
		tor_dirclient::get_resource(request.as_requestable(), dirinfo, &dirmgr.runtime, circmgr)
		.await;
		tor_dirclient::get_resource(request.as_requestable(), dirinfo, rt, circmgr.clone()).await;

		dirmgr.note_request_outcome(&outcome);
		note_request_outcome(&circmgr, &outcome);

		let resource = outcome?;
		Ok((request, resource))
		}

		/// Testing helper: if this is Some, then we return it in place of any
		/// response to fetch_multiple.
		///
		/// Note that only one test uses this: otherwise there would be a race
		/// condition. :p
		#[cfg(test)]
		static CANNED_RESPONSE: Lazy<Mutex<Vec<String>>> = Lazy::new(\|\| Mutex::new(vec![]));

		/// Launch a set of download requests for a set of missing objects in
		/// `missing`, and return each request along with the response it received.
		///
		/// Don't launch more than `parallelism` requests at once.
		async fn fetch_multiple<R: Runtime>(
		dirmgr: Arc<DirMgr<R>>,
		missing: Vec<DocId>,
		missing: &[DocId],
		parallelism: usize,
		) -> Result<Vec<(ClientRequest, DirResponse)>> {
		let mut requests = Vec::new();
		for (_type, query) in docid::partition_by_type(missing.into_iter()) {
		requests.extend(dirmgr.query_into_requests(query)?);
		let requests = {
		let store = dirmgr.store.lock().expect("store lock poisoned");
		make_requests_for_documents(&dirmgr.runtime, missing, store.deref())?
		};

		#[cfg(test)]
		{
		let m = CANNED_RESPONSE.lock().expect("Poisoned mutex");
		if !m.is_empty() {
		return Ok(requests
		.into_iter()
		.zip(m.iter().map(DirResponse::from_body))
		.collect());
		}
		}

		let circmgr = dirmgr.circmgr()?;
		// TODO(eta): Maybe don't hold this netdir for so long?
		let netdir = dirmgr.opt_netdir();

		// TODO: instead of waiting for all the queries to finish, we
		// could stream the responses back or something.
		let responses: Vec<Result<(ClientRequest, DirResponse)>> = futures::stream::iter(requests)
		.map(\|query\| fetch_single(Arc::clone(&dirmgr), query))
		.map(\|query\| fetch_single(&dirmgr.runtime, query, netdir.as_deref(), circmgr.clone()))
		.buffer_unordered(parallelism)
		.collect()
		.await;
		@@ -132,9 +273,13 @@ async fn load_once<R: Runtime>(
		"Found {} missing documents; trying to load them",
		missing.len()
		);
		let documents = load_all(dirmgr, missing)?;

		match state.add_from_cache(documents, dirmgr.store_if_rw()) {
		let documents = {
		let store = dirmgr.store.lock().expect("store lock poisoned");
		load_documents_from_store(&missing, store.deref())?
		};

		match state.add_from_cache(documents) {
		Err(Error::UntimelyObject(TimeValidityError::Expired(_))) => {
		// This is just an expired object from the cache; we don't need
		// to call that an error. Treat it as if it were absent.
		@@ -163,6 +308,10 @@ pub(crate) async fn load<R: Runtime>(
		loop {
		trace!(state=%state.describe(), "Loading from cache");
		let changed = load_once(&dirmgr, &mut state).await?;
		{
		let mut store = dirmgr.store.lock().expect("store lock poisoned");
		dirmgr.apply_netdir_changes(&mut state, store.deref_mut())?;
		}

		if state.can_advance() {
		state = state.advance()?;
		@@ -196,7 +345,7 @@ async fn download_attempt<R: Runtime>(
		) -> Result<bool> {
		let mut changed = false;
		let missing = state.missing_docs();
		let fetched = fetch_multiple(Arc::clone(dirmgr), missing, parallelism).await?;
		let fetched = fetch_multiple(Arc::clone(dirmgr), &missing, parallelism).await?;
		for (client_req, dir_response) in fetched {
		let source = dir_response.source().map(Clone::clone);
		let text = match String::from_utf8(dir_response.into_output())
		@@ -205,7 +354,7 @@ async fn download_attempt<R: Runtime>(
		Ok(t) => t,
		Err(e) => {
		if let Some(source) = source {
		dirmgr.note_cache_error(&source, &e);
		note_cache_error(dirmgr.circmgr()?.deref(), &source, &e);
		}
		continue;
		}
		@@ -217,13 +366,13 @@ async fn download_attempt<R: Runtime>(
		Ok(b) => {
		changed \|= b;
		if let Some(source) = source {
		dirmgr.note_cache_success(&source);
		note_cache_success(dirmgr.circmgr()?.deref(), &source);
		}
		}
		Err(e) => {
		warn!("error while adding directory info: {}", e);
		if let Some(source) = source {
		dirmgr.note_cache_error(&source, &e);
		note_cache_error(dirmgr.circmgr()?.deref(), &source, &e);
		}
		}
		}
		@@ -231,7 +380,7 @@ async fn download_attempt<R: Runtime>(
		Err(e) => {
		warn!("Error when expanding directory text: {}", e);
		if let Some(source) = source {
		dirmgr.note_cache_error(&source, &e);
		note_cache_error(dirmgr.circmgr()?.deref(), &source, &e);
		}
		}
		}
		@@ -266,7 +415,7 @@ pub(crate) async fn download<R: Runtime>(
		let runtime = upgrade_weak_ref(&dirmgr)?.runtime.clone();

		'next_state: loop {
		let retry_config = state.dl_config()?;
		let retry_config = state.dl_config();
		let parallelism = retry_config.parallelism();

		// In theory this could be inside the loop below maybe? If we
		@@ -275,7 +424,7 @@ pub(crate) async fn download<R: Runtime>(
		let mut now = {
		let dirmgr = upgrade_weak_ref(&dirmgr)?;
		load_once(&dirmgr, &mut state).await?;
		dirmgr.now()
		dirmgr.runtime.wallclock()
		};

		// Skip the downloads if we can...
		@@ -283,6 +432,13 @@ pub(crate) async fn download<R: Runtime>(
		state = state.advance()?;
		continue 'next_state;
		}
		// Apply any netdir changes that the state gives us.
		// TODO(eta): Consider deprecating state.is_ready().
		{
		let dirmgr = upgrade_weak_ref(&dirmgr)?;
		let mut store = dirmgr.store.lock().expect("store lock poisoned");
		dirmgr.apply_netdir_changes(&mut state, store.deref_mut())?;
		}
		if state.is_ready(Readiness::Complete) {
		return Ok((state, None));
		}
		@@ -339,9 +495,17 @@ pub(crate) async fn download<R: Runtime>(
		continue 'next_state;
		},
		};
		dirmgr.now()
		dirmgr.runtime.wallclock()
		};

		// Apply any netdir changes that the state gives us.
		// TODO(eta): Consider deprecating state.is_ready().
		{
		let dirmgr = upgrade_weak_ref(&dirmgr)?;
		let mut store = dirmgr.store.lock().expect("store lock poisoned");
		dirmgr.apply_netdir_changes(&mut state, store.deref_mut())?;
		}

		// Exit if there is nothing more to download.
		if state.is_ready(Readiness::Complete) {
		return Ok((state, None));
		@@ -483,11 +647,7 @@ mod test {
		})
		.collect()
		}
		fn add_from_cache(
		&mut self,
		docs: HashMap<DocId, DocumentText>,
		_storage: Option<&Mutex<DynStore>>,
		) -> Result<bool> {
		fn add_from_cache(&mut self, docs: HashMap<DocId, DocumentText>) -> Result<bool> {
		let mut changed = false;
		for id in docs.keys() {
		if let DocId::Microdesc(id) = id {
		@@ -518,8 +678,8 @@ mod test {
		}
		Ok(changed)
		}
		fn dl_config(&self) -> Result<DownloadSchedule> {
		Ok(DownloadSchedule::default())
		fn dl_config(&self) -> DownloadSchedule {
		DownloadSchedule::default()
		}
		fn advance(self: Box<Self>) -> Result<Box<dyn DirState>> {
		if self.can_advance() {
		@@ -584,11 +744,11 @@ mod test {
		{
		let mut resp = CANNED_RESPONSE.lock().unwrap();
		// H4 and H5.
		*resp = Some(
		*resp = vec![
		"7768696c652069206c696b6520746f207761746368207468696e6773206f6e20
		545620536174656c6c697465206f66206c6f766520536174656c6c6974652d2d"
		.to_owned(),
		);
		];
		}
		let mgr = Arc::new(mgr);
		let mut on_usable = None;

crates/tor-dirmgr/src/docid.rs

+59 −0

Original line number	Diff line number	Diff line
		@@ -2,7 +2,10 @@
		//! documents we want and which we have.

		use std::{borrow::Borrow, collections::HashMap};
		use tracing::trace;

		use crate::storage::Store;
		use crate::DocumentText;
		use tor_dirclient::request;
		#[cfg(feature = "routerdesc")]
		use tor_netdoc::doc::routerdesc::RdDigest;
		@@ -195,6 +198,62 @@ impl DocQuery {
		}
		}
		}

		/// Load documents specified by this `DocQuery` from the store, if they can be found.
		///
		/// # Note
		///
		/// This function may not return all documents that the query asked for. If this happens, no
		/// error will be returned. It is the caller's responsibility to handle this case.
		pub(crate) fn load_from_store_into(
		&self,
		result: &mut HashMap<DocId, DocumentText>,
		store: &dyn Store,
		) -> crate::Result<()> {
		use DocQuery::*;
		match self {
		LatestConsensus {
		flavor,
		cache_usage,
		} => {
		if *cache_usage == CacheUsage::MustDownload {
		// Do nothing: we don't want a cached consensus.
		trace!("MustDownload is set; not checking for cached consensus.");
		} else if let Some(c) =
		store.latest_consensus(*flavor, cache_usage.pending_requirement())?
		{
		trace!("Found a reasonable consensus in the cache");
		let id = DocId::LatestConsensus {
		flavor: *flavor,
		cache_usage: *cache_usage,
		};
		result.insert(id, c.into());
		}
		}
		AuthCert(ids) => result.extend(
		store
		.authcerts(ids)?
		.into_iter()
		.map(\|(id, c)\| (DocId::AuthCert(id), DocumentText::from_string(c))),
		),
		Microdesc(digests) => {
		result.extend(
		store
		.microdescs(digests)?
		.into_iter()
		.map(\|(id, md)\| (DocId::Microdesc(id), DocumentText::from_string(md))),
		);
		}
		#[cfg(feature = "routerdesc")]
		RouterDesc(digests) => result.extend(
		store
		.routerdescs(digests)?
		.into_iter()
		.map(\|(id, rd)\| (DocId::RouterDesc(id), DocumentText::from_string(rd))),
		),
		}
		Ok(())
		}
		}

		impl From<DocId> for DocQuery {

crates/tor-dirmgr/src/err.rs

+5 −0

Original line number	Diff line number	Diff line
		@@ -14,6 +14,9 @@ pub enum Error {
		/// We received a document we didn't want at all.
		#[error("unwanted object: {0}")]
		Unwanted(&'static str),
		/// The NetDir we downloaded is older than the one we already have.
		#[error("downloaded netdir is older than the one we have")]
		NetDirOlder,
		/// This DirMgr doesn't support downloads.
		#[error("tried to download information on a DirMgr with no download support")]
		NoDownloadSupport,
		@@ -169,6 +172,7 @@ impl Error {
		\| Error::BadHexInCache(_)
		\| Error::OfflineMode
		\| Error::Spawn { .. }
		\| Error::NetDirOlder
		\| Error::Bug(_) => false,

		// For this one, we delegate.
		@@ -224,6 +228,7 @@ impl HasKind for Error {
		E::UnrecognizedSchema => EK::CacheCorrupted,
		E::BadNetworkConfig(_) => EK::InvalidConfig,
		E::DirectoryNotPresent => EK::DirectoryExpired,
		E::NetDirOlder => EK::TorDirectoryError,
		E::BadUtf8FromDirectory(_) => EK::TorProtocolViolation,
		E::BadUtf8InCache(_) => EK::CacheCorrupted,
		E::BadHexInCache(_) => EK::CacheCorrupted,