Merge branch 'dir-filter' into 'main'

arti-client, dirmgr: Initial DirFilter code

See merge request tpo/core/arti!431

crates/arti-client/Cargo.toml

@@ -6,20 +6,22 @@ edition = "2018"
 license = "MIT OR Apache-2.0"
 homepage = "https://gitlab.torproject.org/tpo/core/arti/-/wikis/home"
 description = "Library for connecting to the Tor network as an anonymous client"
-keywords = [ "tor", "arti", "privacy", "anonymity", "networking" ]
-categories = [ "network-programming", "cryptography" ]
-repository="https://gitlab.torproject.org/tpo/core/arti.git/"
+keywords = ["tor", "arti", "privacy", "anonymity", "networking"]
+categories = ["network-programming", "cryptography"]
+repository = "https://gitlab.torproject.org/tpo/core/arti.git/"
 
 [features]
-default = [ "tokio", "native-tls" ]
-async-std = [ "tor-rtcompat/async-std" ]
-tokio = [ "tor-rtcompat/tokio", "tor-proto/tokio" ]
-native-tls = [ "tor-rtcompat/native-tls" ]
-rustls = [ "tor-rtcompat/rustls" ]
-static = [ "static-sqlite", "static-native-tls" ]
-static-sqlite = [ "tor-dirmgr/static" ]
-static-native-tls = [ "tor-rtcompat/static", "native-tls" ]
-error_detail = [ ]
+default = ["tokio", "native-tls"]
+async-std = ["tor-rtcompat/async-std"]
+tokio = ["tor-rtcompat/tokio", "tor-proto/tokio"]
+native-tls = ["tor-rtcompat/native-tls"]
+rustls = ["tor-rtcompat/rustls"]
+static = ["static-sqlite", "static-native-tls"]
+static-sqlite = ["tor-dirmgr/static"]
+static-native-tls = ["tor-rtcompat/static", "native-tls"]
+
+dirfilter = ["tor-dirmgr/dirfilter"]
+error_detail = []
 
 # Enable experimental APIs that are not yet officially supported.
 #
@@ -28,16 +30,16 @@ error_detail = [ ]
 experimental-api = []
 
 [dependencies]
-tor-basic-utils = { path="../tor-basic-utils", version = "0.1.0"}
-tor-circmgr = { path="../tor-circmgr", version = "0.1.0"}
-tor-config = { path="../tor-config", version = "0.1.0"}
-tor-chanmgr = { path="../tor-chanmgr", version = "0.1.0"}
-tor-dirmgr = { path="../tor-dirmgr", version = "0.1.0"}
-tor-error = { path="../tor-error", version = "0.1.0"}
-tor-netdoc = { path="../tor-netdoc", version = "0.1.0"}
-tor-persist = { path="../tor-persist", version = "0.1.0"}
-tor-proto = { path="../tor-proto", version = "0.1.0"}
-tor-rtcompat = { path="../tor-rtcompat", version = "0.1.0"}
+tor-basic-utils = { path = "../tor-basic-utils", version = "0.1.0" }
+tor-circmgr = { path = "../tor-circmgr", version = "0.1.0" }
+tor-config = { path = "../tor-config", version = "0.1.0" }
+tor-chanmgr = { path = "../tor-chanmgr", version = "0.1.0" }
+tor-dirmgr = { path = "../tor-dirmgr", version = "0.1.0" }
+tor-error = { path = "../tor-error", version = "0.1.0" }
+tor-netdoc = { path = "../tor-netdoc", version = "0.1.0" }
+tor-persist = { path = "../tor-persist", version = "0.1.0" }
+tor-proto = { path = "../tor-proto", version = "0.1.0" }
+tor-rtcompat = { path = "../tor-rtcompat", version = "0.1.0" }
 
 humantime-serde = "1.1.1"
 derive_builder = "0.11"
@@ -45,15 +47,27 @@ derive_more = "0.99"
 directories = "4"
 educe = "0.4.6"
 futures = "0.3.14"
-postage = { version = "0.4", default-features = false, features = ["futures-traits"] }
+postage = { version = "0.4", default-features = false, features = [
+    "futures-traits",
+] }
 tracing = "0.1.18"
 serde = { version = "1.0.103", features = ["derive"] }
 thiserror = "1"
 pin-project = "1"
 
 [dev-dependencies]
-tor-rtcompat = { path="../tor-rtcompat", version = "0.1.0", features=["tokio", "native-tls" ] }
-tokio-crate = { package = "tokio", version = "1.7", features = ["rt", "rt-multi-thread", "io-util", "net", "time", "macros" ] }
+tor-rtcompat = { path = "../tor-rtcompat", version = "0.1.0", features = [
+    "tokio",
+    "native-tls",
+] }
+tokio-crate = { package = "tokio", version = "1.7", features = [
+    "rt",
+    "rt-multi-thread",
+    "io-util",
+    "net",
+    "time",
+    "macros",
+] }
 pin-project = "1"
 tokio-util = { version = "0.7.0", features = ["compat"] }
 anyhow = "1.0.23"

crates/arti-client/src/builder.rs

@@ -56,6 +56,11 @@ pub struct TorClientBuilder<R: Runtime> {
     /// Wrapped in an Arc so that we don't need to force DirProviderBuilder to
     /// implement Clone.
     dirmgr_builder: Arc<dyn DirProviderBuilder<R>>,
+    /// Optional directory filter to install for testing purposes.
+    ///
+    /// Only available when `arti-client` is built with the `dirfilter` and `experimental-api` features.
+    #[cfg(feature = "dirfilter")]
+    dirfilter: tor_dirmgr::filter::FilterConfig,
 }
 
 impl<R: Runtime> TorClientBuilder<R> {
@@ -66,6 +71,8 @@ impl<R: Runtime> TorClientBuilder<R> {
             config: TorClientConfig::default(),
             bootstrap_behavior: BootstrapBehavior::default(),
             dirmgr_builder: Arc::new(DirMgrBuilder {}),
+            #[cfg(feature = "dirfilter")]
+            dirfilter: None,
         }
     }
 
@@ -99,6 +106,19 @@ impl<R: Runtime> TorClientBuilder<R> {
         self
     }
 
+    /// Install a [`DirFilter`](tor_dirmgr::filter::DirFilter) to
+    ///
+    /// Only available whe compiled with the the `dirfilter` feature: this code
+    /// is unstable and not recommended for production use.
+    #[cfg(feature = "dirfilter")]
+    pub fn dirfilter<F>(mut self, filter: F) -> Self
+    where
+        F: Into<Arc<dyn tor_dirmgr::filter::DirFilter + 'static>>,
+    {
+        self.dirfilter = Some(filter.into());
+        self
+    }
+
     /// Create a `TorClient` from this builder, without automatically launching
     /// the bootstrap process.
     ///
@@ -115,11 +135,19 @@ impl<R: Runtime> TorClientBuilder<R> {
     /// process (for example, you might wish to avoid initiating network
     /// connections until explicit user confirmation is given).
     pub fn create_unbootstrapped(self) -> Result<TorClient<R>> {
+        #[allow(unused_mut)]
+        let mut dirmgr_extensions = tor_dirmgr::config::DirMgrExtensions::default();
+        #[cfg(feature = "dirfilter")]
+        {
+            dirmgr_extensions.filter = self.dirfilter;
+        }
+
         TorClient::create_inner(
             self.runtime,
             self.config,
             self.bootstrap_behavior,
             self.dirmgr_builder.as_ref(),
+            dirmgr_extensions,
         )
         .map_err(ErrorDetail::into)
     }

crates/arti-client/src/client.rs

@@ -354,8 +354,13 @@ impl<R: Runtime> TorClient<R> {
         config: TorClientConfig,
         autobootstrap: BootstrapBehavior,
         dirmgr_builder: &dyn crate::builder::DirProviderBuilder<R>,
+        dirmgr_extensions: tor_dirmgr::config::DirMgrExtensions,
     ) -> StdResult<Self, ErrorDetail> {
-        let dir_cfg = (&config).try_into()?;
+        let dir_cfg = {
+            let mut c: tor_dirmgr::DirMgrConfig = (&config).try_into()?;
+            c.extensions = dirmgr_extensions;
+            c
+        };
         let statemgr = FsStateMgr::from_path(config.storage.expand_state_dir()?)?;
         let addr_cfg = config.address_filter.clone();
 

crates/arti-client/src/config.rs

@@ -300,6 +300,7 @@ impl TryInto<dir::DirMgrConfig> for &TorClientConfig {
             schedule_config:     self.download_schedule  .clone(),
             cache_path:          self.storage.expand_cache_dir()?,
             override_net_params: self.override_net_params.clone(),
+            extensions:          Default::default(),
         })
     }
 }

crates/tor-dirmgr/Cargo.toml

@@ -16,6 +16,7 @@ mmap = ["memmap2"]
 static = ["rusqlite/bundled"]
 # (Incomplete) support for downloading and storing router descriptors
 routerdesc = ["tor-dirclient/routerdesc"]
+dirfilter = []
 
 # Enable experimental APIs that are not yet officially supported.
 #
@@ -24,18 +25,18 @@ routerdesc = ["tor-dirclient/routerdesc"]
 experimental-api = []
 
 [dependencies]
-tor-basic-utils = { path="../tor-basic-utils", version = "0.1.0"}
-retry-error = { path = "../retry-error", version = "0.1.0"}
-tor-checkable = { path = "../tor-checkable", version = "0.1.0"}
-tor-circmgr = { path = "../tor-circmgr", version = "0.1.0"}
-tor-config = { path = "../tor-config", version = "0.1.0"}
-tor-consdiff = { path = "../tor-consdiff", version = "0.1.0"}
-tor-dirclient = { path = "../tor-dirclient", version = "0.1.0"}
-tor-error = { path="../tor-error", version = "0.1.0"}
-tor-netdir = { path = "../tor-netdir", version = "0.1.0"}
-tor-netdoc = { path = "../tor-netdoc", version = "0.1.0"}
-tor-llcrypto = { path = "../tor-llcrypto", version = "0.1.0"}
-tor-rtcompat = { path = "../tor-rtcompat", version = "0.1.0"}
+tor-basic-utils = { path = "../tor-basic-utils", version = "0.1.0" }
+retry-error = { path = "../retry-error", version = "0.1.0" }
+tor-checkable = { path = "../tor-checkable", version = "0.1.0" }
+tor-circmgr = { path = "../tor-circmgr", version = "0.1.0" }
+tor-config = { path = "../tor-config", version = "0.1.0" }
+tor-consdiff = { path = "../tor-consdiff", version = "0.1.0" }
+tor-dirclient = { path = "../tor-dirclient", version = "0.1.0" }
+tor-error = { path = "../tor-error", version = "0.1.0" }
+tor-netdir = { path = "../tor-netdir", version = "0.1.0" }
+tor-netdoc = { path = "../tor-netdoc", version = "0.1.0" }
+tor-llcrypto = { path = "../tor-llcrypto", version = "0.1.0" }
+tor-rtcompat = { path = "../tor-rtcompat", version = "0.1.0" }
 
 async-trait = "0.1.2"
 base64 = "0.13.0"
@@ -66,5 +67,8 @@ humantime-serde = "1.1.1"
 futures-await-test = "0.3.0"
 hex-literal = "0.3"
 tempfile = "3"
-tor-rtcompat = { path = "../tor-rtcompat", version = "0.1.0", features = [ "tokio", "native-tls" ] }
+tor-rtcompat = { path = "../tor-rtcompat", version = "0.1.0", features = [
+    "tokio",
+    "native-tls",
+] }
 float_eq = "0.7"

crates/tor-dirmgr/src/config.rs

@@ -199,6 +199,12 @@ pub struct DirMgrConfig {
     /// immediately. Users should _not_ assume that the effect of changing this
     /// option will always be delayed.)
     pub override_net_params: netstatus::NetParams<i32>,
+
+    /// Extra fields for extension purposes.
+    ///
+    /// These are kept in a separate type so that the type can be marked as
+    /// `non_exhaustive` and used for optional features.
+    pub extensions: DirMgrExtensions,
 }
 
 impl DirMgrConfig {
@@ -252,6 +258,7 @@ impl DirMgrConfig {
             },
             schedule_config: new_config.schedule_config.clone(),
             override_net_params: new_config.override_net_params.clone(),
+            extensions: new_config.extensions.clone(),
         }
     }
 
@@ -265,6 +272,15 @@ impl DirMgrConfig {
     }
 }
 
+/// Optional extensions for configuring
+#[derive(Debug, Clone, Default)]
+#[non_exhaustive]
+pub struct DirMgrExtensions {
+    /// A filter to be used when installing new directory objects.
+    #[cfg(feature = "dirfilter")]
+    pub filter: crate::filter::FilterConfig,
+}
+
 impl DownloadScheduleConfig {
     /// Return configuration for retrying our entire bootstrap
     /// operation at startup.

crates/tor-dirmgr/src/filter.rs

+//! A filtering mechanism for directory objects.
+//!
+//! This module and its members are only available when `tor-dirmgr` is built
+//! with the `dirfilter` feature.
+//!
+//! This is unstable code, currently used for testing only.  It might go away in
+//! future versions, or its API might change completely. There are no semver
+//! guarantees.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use crate::Result;
+use tor_netdoc::doc::{microdesc::Microdesc, netstatus::UncheckedMdConsensus};
+
+/// Filtering configuration, as provided to the directory code
+pub type FilterConfig = Option<Arc<dyn DirFilter>>;
+
+/// An object that can filter directory documents before they're handled.
+///
+/// Instances of DirFilter can be used for testing, to modify directory data
+/// on-the-fly.
+pub trait DirFilter: Debug + Send + Sync {
+    /// Modify `consensus` in an unspecified way.
+    fn filter_consensus(&self, consensus: UncheckedMdConsensus) -> Result<UncheckedMdConsensus> {
+        Ok(consensus)
+    }
+    /// Modify `md` in an unspecified way.
+    fn filter_md(&self, md: Microdesc) -> Result<Microdesc> {
+        Ok(md)
+    }
+}
+
+/// A [`DirFilter`] that does nothing.
+#[derive(Debug)]
+#[allow(clippy::exhaustive_structs)]
+pub struct NilFilter;
+
+impl DirFilter for NilFilter {}

crates/tor-dirmgr/src/lib.rs

@@ -66,6 +66,9 @@ mod shared_ref;
 mod state;
 mod storage;
 
+#[cfg(feature = "dirfilter")]
+pub mod filter;
+
 use crate::docid::{CacheUsage, ClientRequest, DocQuery};
 #[cfg(not(feature = "experimental-api"))]
 use crate::shared_ref::SharedMutArc;
@@ -222,6 +225,10 @@ pub struct DirMgr<R: Runtime> {
     ///
     /// (In offline mode, this does nothing.)
     bootstrap_started: AtomicBool,
+
+    /// A filter that gets applied to directory objects before we use them.
+    #[cfg(feature = "dirfilter")]
+    filter: crate::filter::FilterConfig,
 }
 
 /// RAII guard to reset an AtomicBool on drop.
@@ -702,6 +709,8 @@ impl<R: Runtime> DirMgr<R> {
         let receive_status = DirBootstrapEvents {
             inner: receive_status,
         };
+        #[cfg(feature = "dirfilter")]
+        let filter = config.extensions.filter.clone();
 
         Ok(DirMgr {
             config: config.into(),
@@ -714,6 +723,8 @@ impl<R: Runtime> DirMgr<R> {
             runtime,
             offline,
             bootstrap_started: AtomicBool::new(false),
+            #[cfg(feature = "dirfilter")]
+            filter,
         })
     }
 

crates/tor-dirmgr/src/state.rs

@@ -84,6 +84,10 @@ pub(crate) trait WriteNetDir: 'static + Sync + Send {
     /// testing it is helpful to be able to mock our our current view
     /// of the time.
     fn now(&self) -> SystemTime;
+
+    /// Return the currently configured DynFilter for this state.
+    #[cfg(feature = "dirfilter")]
+    fn filter(&self) -> &dyn crate::filter::DirFilter;
 }
 
 impl<R: Runtime> WriteNetDir for crate::DirMgr<R> {
@@ -108,6 +112,11 @@ impl<R: Runtime> WriteNetDir for crate::DirMgr<R> {
     fn now(&self) -> SystemTime {
         SystemTime::now()
     }
+
+    #[cfg(feature = "dirfilter")]
+    fn filter(&self) -> &dyn crate::filter::DirFilter {
+        self.filter.as_deref().unwrap_or(&crate::filter::NilFilter)
+    }
 }
 
 /// Initial state: fetching or loading a consensus directory.
@@ -278,6 +287,12 @@ impl<DM: WriteNetDir> GetConsensusState<DM> {
         let (consensus_meta, unvalidated) = {
             let (signedval, remainder, parsed) =
                 MdConsensus::parse(text).map_err(|e| Error::from_netdoc(source.clone(), e))?;
+            #[cfg(feature = "dirfilter")]
+            let parsed = if let Some(wd) = Weak::upgrade(&self.writedir) {
+                wd.filter().filter_consensus(parsed)?
+            } else {
+                parsed
+            };
             let now = current_time(&self.writedir)?;
             if let Ok(timely) = parsed.check_valid_at(&now) {
                 let meta = ConsensusMeta::from_unvalidated(signedval, remainder, &timely);
@@ -651,6 +666,14 @@ impl<DM: WriteNetDir> GetMicrodescsState<DM> {
     where
         I: IntoIterator<Item = Microdesc>,
     {
+        #[cfg(feature = "dirfilter")]
+        let mds: Vec<Microdesc> = if let Some(wd) = Weak::upgrade(&self.writedir) {
+            mds.into_iter()
+                .filter_map(|m| wd.filter().filter_md(m).ok())
+                .collect()
+        } else {
+            mds.into_iter().collect()
+        };
         if let Some(p) = &mut self.partial {
             for md in mds {
                 self.newly_listed.push(*md.digest());
@@ -1033,6 +1056,10 @@ mod test {
         fn now(&self) -> SystemTime {
             self.now
         }
+        #[cfg(feature = "dirfilter")]
+        fn filter(&self) -> &dyn crate::filter::DirFilter {
+            &crate::filter::NilFilter
+        }
     }
 
     // Test data