From e7fcaa980f2dc4385ed6d3acc2cf48a2be54b373 Mon Sep 17 00:00:00 2001 From: Jack Wampler Date: Wed, 11 Feb 2026 07:04:53 -0700 Subject: [PATCH] HTTP & DNS Improvements (#6423) * Improve HTTP use of connection pooling (#6375) * add swap to system resolver instead of fallback (#6376) * add header tracking outer host name used in stealth requests (#6389) * Rotate urls on parse failure (#6383) * Add shared settings for stealth policy across HTTP clients (#6388) * Better controls for global interaction w/ static DNS (#6374) --- common/http-api-client/src/dns.rs | 296 +++++++++++----- common/http-api-client/src/dns/constants.rs | 6 + .../src/dns/static_resolver.rs | 331 +++++++++++++++--- common/http-api-client/src/fronted.rs | 233 +++++++++++- common/http-api-client/src/lib.rs | 269 +++++++++----- common/http-api-client/src/tests.rs | 5 +- common/http-api-client/src/url.rs | 10 + 7 files changed, 908 insertions(+), 242 deletions(-) diff --git a/common/http-api-client/src/dns.rs b/common/http-api-client/src/dns.rs index 0d61c75362..b055dbfd0c 100644 --- a/common/http-api-client/src/dns.rs +++ b/common/http-api-client/src/dns.rs @@ -46,7 +46,10 @@ use std::{ collections::HashMap, net::{IpAddr, SocketAddr}, str::FromStr, - sync::{Arc, LazyLock}, + sync::{ + Arc, LazyLock, + atomic::{AtomicBool, Ordering::Relaxed}, + }, time::Duration, }; @@ -70,14 +73,23 @@ pub(crate) const DEFAULT_QUERY_TIMEOUT: Duration = Duration::from_secs(5); impl ClientBuilder { /// Override the DNS resolver implementation used by the underlying http client. + /// This forces the use of an independent request executor (via [`Self::non_shared`]). pub fn dns_resolver(mut self, resolver: Arc) -> Self { - self.reqwest_client_builder = self.reqwest_client_builder.dns_resolver(resolver); + self = self.non_shared(); + // because of the call to non-shared this conditional should always run. + if let Some(rb) = self.reqwest_client_builder { + self.reqwest_client_builder = Some(rb.dns_resolver(resolver)); + } self.use_secure_dns = false; self } - /// Override the DNS resolver implementation used by the underlying http client. + /// Override the DNS resolver implementation used by the underlying http client. If + /// [`Self::dns_resolver`] is called directly that will take priority over this, there is no + /// need to call both. + /// This forces the use of an independent request executor (via [`Self::non_shared`]). pub fn no_hickory_dns(mut self) -> Self { + self = self.non_shared(); self.use_secure_dns = false; self } @@ -129,7 +141,8 @@ pub struct HickoryDnsResolver { // Tokio Runtime in initialization, so we must delay the actual // construction of the resolver. state: Arc>, - fallback: Option>>, + use_system: Arc, + system_resolver: Arc>, static_base: Option>>, use_shared: bool, /// Overall timeout for dns lookup associated with any individual host resolution. For example, @@ -141,7 +154,8 @@ impl Default for HickoryDnsResolver { fn default() -> Self { Self { state: Default::default(), - fallback: Default::default(), + use_system: Arc::new(AtomicBool::new(false)), + system_resolver: Default::default(), static_base: Some(Default::default()), use_shared: true, overall_dns_timeout: DEFAULT_OVERALL_LOOKUP_TIMEOUT, @@ -151,16 +165,28 @@ impl Default for HickoryDnsResolver { impl Resolve for HickoryDnsResolver { fn resolve(&self, name: Name) -> Resolving { - let resolver = self.state.clone(); - let maybe_fallback = self.fallback.clone(); - let maybe_static = self.static_base.clone(); + let use_system = self.use_system.load(std::sync::atomic::Ordering::Relaxed); let use_shared = self.use_shared; + let resolver = if use_system { + match self + .system_resolver + .get_or_try_init(|| HickoryDnsResolver::new_resolver_system(use_shared)) + { + Ok(r) => r.clone(), + Err(e) => return Box::pin(return_err(e)), + } + } else { + self.state + .get_or_init(|| HickoryDnsResolver::new_resolver(use_shared)) + .clone() + }; + + let maybe_static = self.static_base.clone(); let overall_dns_timeout = self.overall_dns_timeout; Box::pin(async move { resolve( name, resolver, - maybe_fallback, maybe_static, use_shared, overall_dns_timeout, @@ -171,16 +197,17 @@ impl Resolve for HickoryDnsResolver { } } +async fn return_err(e: ResolveError) -> Result> { + Err(Box::new(e) as Box) +} + async fn resolve( name: Name, - resolver: Arc>, - maybe_fallback: Option>>, + resolver: TokioResolver, maybe_static: Option>>, independent: bool, overall_dns_timeout: Duration, ) -> Result { - let resolver = resolver.get_or_init(|| HickoryDnsResolver::new_resolver(independent)); - // try checking the static table to see if any of the addresses in the table have been // looked up previously within the timeout to where we are not yet ready to try the // default resolver yet again. @@ -214,22 +241,6 @@ async fn resolve( } }; - // If the primary resolver encountered an error, attempt a lookup using the fallback - // resolver if one is configured. - if let Some(ref fallback) = maybe_fallback { - let resolver = - fallback.get_or_try_init(|| HickoryDnsResolver::new_resolver_system(independent))?; - - let resolve_fut = - tokio::time::timeout(overall_dns_timeout, resolver.lookup_ip(name.as_str())); - if let Ok(Ok(lookup)) = resolve_fut.await { - let addrs: Addrs = Box::new(SocketAddrs { - iter: lookup.into_iter(), - }); - return Ok(addrs); - } - } - // If no record has been found and a static map of fallback addresses is configured // check the table for our entry if let Some(ref static_resolver) = maybe_static { @@ -258,6 +269,11 @@ impl Iterator for SocketAddrs { } impl HickoryDnsResolver { + /// Returns an instance of the shared resolver. + pub fn shared() -> Self { + SHARED_RESOLVER.clone() + } + /// Attempt to resolve a domain name to a set of ['IpAddr']s pub async fn resolve_str( &self, @@ -265,10 +281,20 @@ impl HickoryDnsResolver { ) -> Result + use<>, ResolveError> { let n = Name::from_str(name).map_err(|_| ResolveError::InvalidNameError(name.to_string()))?; + let use_system = self.use_system.load(std::sync::atomic::Ordering::Relaxed); + let resolver = if use_system { + self.system_resolver + .get_or_try_init(|| HickoryDnsResolver::new_resolver_system(self.use_shared))? + .clone() + } else { + self.state + .get_or_init(|| HickoryDnsResolver::new_resolver(self.use_shared)) + .clone() + }; + resolve( n, - self.state.clone(), - self.fallback.clone(), + resolver, self.static_base.clone(), self.use_shared, self.overall_dns_timeout, @@ -298,13 +324,11 @@ impl HickoryDnsResolver { fn new_resolver_system(use_shared: bool) -> Result { // using a closure here is slightly gross, but this makes sure that if the // lazy-init returns an error it can be handled by the client - if !use_shared || SHARED_RESOLVER.fallback.is_none() { + if !use_shared { new_resolver_system() } else { Ok(SHARED_RESOLVER - .fallback - .as_ref() - .unwrap() + .system_resolver .get_or_try_init(new_resolver_system)? .clone()) } @@ -320,45 +344,80 @@ impl HickoryDnsResolver { } } - /// Enable fallback to the system default resolver if the primary (DoX) resolver fails - pub fn enable_system_fallback(&mut self) -> Result<(), ResolveError> { - self.fallback = Some(Default::default()); - let _ = self - .fallback - .as_ref() - .unwrap() - .get_or_try_init(new_resolver_system)?; + /// Swap the primary internal resolver to the system resolver rather than the + /// configured custom resolver. + pub fn use_system_resolver(&self) { + self.use_system.store(true, Relaxed); - // IF THIS INSTANCE IS A FRONT FOR THE SHARED RESOLVER SHOULDN'T THIS FN ENABLE THE SYSTEM FALLBACK FOR THE SHARED RESOLVER TOO? - // if self.use_shared { - // SHARED_RESOLVER.enable_system_fallback()?; - // } - Ok(()) + if self.use_shared { + SHARED_RESOLVER.use_system_resolver(); + } } - /// Disable fallback resolution. If the primary resolver fails the error is - /// returned immediately - pub fn disable_system_fallback(&mut self) { - self.fallback = None; + /// Swap the primary internal resolver to the configured custom resolver rather than the + /// system resolver. + pub fn use_configured_resolver(&self) { + self.use_system.store(false, Relaxed); - // // IF THIS INSTANCE IS A FRONT FOR THE SHARED RESOLVER SHOULDN'T THIS FN ENABLE THE SYSTEM FALLBACK FOR THE SHARED RESOLVER TOO? - // if self.use_shared { - // SHARED_RESOLVER.fallback = None; - // } + if self.use_shared { + SHARED_RESOLVER.use_configured_resolver(); + } } - /// Get the current map of hostname to address in use by the fallback static lookup if one + /// Clear entries from the static table that would return entries during the pre-resolve stage. + /// This means that all lookups will attempt to use the network resolver again before the static + /// table is consulted. + /// + /// Entries elevated to pre-resolve from fallback (added from default or using + /// [`set_fallback`]`) will have their cache timeout cleared. Entries added directly to + /// pre-resolve (using [`Self::set_static_preresolve`]) will be removed. + pub fn clear_preresolve(&self) { + debug!("clearing pre-resolve table"); + if let Some(cell) = &self.static_base + && let Some(static_base) = cell.get() + { + static_base.clear_preresolve() + } + } + + /// Get the current map of hostnames to addresses used in the fallback static lookup stage if one /// exists. pub fn get_static_fallbacks(&self) -> Option>> { - Some(self.static_base.as_ref()?.get()?.get_addrs()) + Some(self.static_base.as_ref()?.get()?.get_fallback_addrs()) } /// Set (or overwrite) the map of addresses used in the fallback static hostname lookup - pub fn set_static_fallbacks(&mut self, addrs: HashMap>) { - let cell = OnceCell::new(); - cell.set(StaticResolver::new(addrs)) - .expect("infallible assign"); - self.static_base = Some(Arc::new(cell)); + pub fn set_fallback_addrs(&mut self, addrs: HashMap>) { + debug!("setting fallback entries for {:?}", addrs.keys()); + if self.static_base.is_none() { + let cell = OnceCell::new(); + self.static_base = Some(Arc::new(cell)); + } + self.static_base + .as_ref() + .unwrap() + .get_or_init(|| Self::new_static_fallback(self.use_shared)) + .set_fallback(addrs); + } + + /// Get the current map of hostnames to addresses used in the preresolve static lookup stage + /// if one exists. + pub fn get_static_preresolve(&self) -> Option>> { + Some(self.static_base.as_ref()?.get()?.get_preresolve_addrs()) + } + + /// Set (or overwrite) the map of addresses used in the preresolve static hostname lookup + pub fn set_static_preresolve(&mut self, addrs: HashMap>) { + debug!("setting pre-resolve entries for {:?}", addrs.keys()); + if self.static_base.is_none() { + let cell = OnceCell::new(); + self.static_base = Some(Arc::new(cell)); + } + self.static_base + .as_ref() + .unwrap() + .get_or_init(|| Self::new_static_fallback(self.use_shared)) + .set_preresolve(addrs); } /// Successfully resolved addresses are cached for a minimum of 30 minutes @@ -495,7 +554,7 @@ fn new_resolver_system() -> Result { } fn new_default_static_fallback() -> StaticResolver { - StaticResolver::new(constants::default_static_addrs()) + StaticResolver::new().with_fallback(constants::default_static_addrs()) } /// Do a trial resolution using each nameserver individually to test which are working and which @@ -532,10 +591,7 @@ mod test { use super::*; use itertools::Itertools; use std::collections::HashMap; - use std::{ - net::{IpAddr, Ipv4Addr, Ipv6Addr}, - time::Instant, - }; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; /// IP addresses guaranteed to fail attempts to resolve /// @@ -597,7 +653,7 @@ mod test { let example_ip6: IpAddr = "dead::beef".parse().unwrap(); addr_map.insert(example_domain.to_string(), vec![example_ip4, example_ip6]); - resolver.set_static_fallbacks(addr_map); + resolver.set_fallback_addrs(addr_map); let mut addrs = resolver.resolve_str(example_domain).await?; assert!(addrs.contains(&example_ip4)); @@ -738,18 +794,19 @@ mod test { } #[tokio::test] - #[ignore] - // this test is dependent of external network setup -- i.e. blocking all traffic to the default - // resolvers. Otherwise the default resolvers will succeed without using the static fallback, - // making the test pointless + #[cfg(any())] // #[ignore] we run --ignore in CI/CD assuming it just means slow -_- + // This test impacts the state of the shared resolver and as such is disabled to avoid + // interference with other tests. + // + // this test is dependent of external network setup -- i.e. blocking all traffic to the + // default resolvers. Otherwise the default resolvers will succeed without using the static + // fallback, making the test pointless async fn dns_lookup_failure_on_shared() -> Result<(), ResolveError> { - let time_start = Instant::now(); - let r = OnceCell::new(); - r.set(build_broken_resolver().expect("failed to build resolver")) - .expect("broken resolver init error"); + let resolver1 = HickoryDnsResolver::shared(); - // create a new resolver that won't mess with the shared resolver used by other tests - let resolver = HickoryDnsResolver::default(); + let time_start = std::time::Instant::now(); + // create a new resolver that uses the shared resolver + let resolver = HickoryDnsResolver::shared(); // successful lookup using fallback to static resolver let domain = "rpc.nymtech.net"; @@ -758,9 +815,27 @@ mod test { .await .expect("failed to resolve address in static lookup"); - println!( - "{}ms resolved {domain}", - (Instant::now() - time_start).as_millis() + let lookup_dur = Instant::now() - time_start; + assert!( + lookup_dur > resolver.overall_dns_timeout, + "expected lookup timeout - took {}ms", + (lookup_dur).as_millis() + ); + + let time_start = std::time::Instant::now(); + // successful lookup using pre-resolve entry promoted from fallback + let domain = "rpc.nymtech.net"; + let _ = resolver1 + .resolve_str(domain) + .await + .expect("domain expected to be in pre-resolve"); + + // this lookup should basically be instant as we are using pre-resolve + let lookup_dur = std::time::Instant::now() - time_start; + assert!( + lookup_dur < Duration::from_millis(10), + "expected instant - took {}ms", + (lookup_dur).as_millis() ); // unsuccessful lookup - primary times out, and not in static table @@ -771,5 +846,62 @@ mod test { // assert!(result.is_err_and(|e| matches!(e, ResolveError::ResolveError(e) if e.is_nx_domain()))); Ok(()) } + + #[tokio::test] + #[cfg(any())] // #[ignore] we run --ignore in CI/CD assuming it just means slow -_- + // This test impacts the state of the shared resolver and as such is disabled to avoid + // interference with other tests. + async fn setting_dns_fallbacks_with_shared_resolver() -> Result<(), ResolveError> { + let resolver1 = HickoryDnsResolver::shared(); + + // create a new resolver that uses the shared resolver + let mut resolver = HickoryDnsResolver::shared(); + + let example_domains = [ + String::from("static1.nymvpn.com"), + String::from("static2.nymvpn.com"), + ]; + let mut addr_map1 = HashMap::new(); + addr_map1.insert( + example_domains[0].clone(), + vec![Ipv4Addr::new(10, 10, 10, 10).into()], + ); + addr_map1.insert( + example_domains[1].clone(), + vec![Ipv4Addr::new(1, 1, 1, 1).into()], + ); + + resolver.set_static_preresolve(addr_map1); + + let time_start = std::time::Instant::now(); + // successful lookup using pre-resolve entry promoted from fallback + let _ = resolver1 + .resolve_str(&example_domains[0]) + .await + .expect("domain expected to be in pre-resolve"); + + // this lookup should basically be instant as we are using pre-resolve + let lookup_dur = std::time::Instant::now() - time_start; + assert!( + lookup_dur < Duration::from_millis(10), + "expected instant - took {}ms", + (lookup_dur).as_millis() + ); + + // After clearing the pre-resolve in one instance of the shared resolver ... + resolver.clear_preresolve(); + + // ... other instances have their pre-resolve entries cleared. + let prereslve_lookup = resolver1 + .static_base + .as_ref() + .unwrap() + .get() + .unwrap() + .pre_resolve(&example_domains[0]); + assert!(prereslve_lookup.is_none()); + + Ok(()) + } } } diff --git a/common/http-api-client/src/dns/constants.rs b/common/http-api-client/src/dns/constants.rs index b13d6f3fde..5a7c35037f 100644 --- a/common/http-api-client/src/dns/constants.rs +++ b/common/http-api-client/src/dns/constants.rs @@ -72,6 +72,12 @@ pub const NYM_RPC_IPS: &[IpAddr] = &[ )), ]; +#[allow(unused)] +pub fn empty_static_addrs() -> HashMap> { + HashMap::new() +} + +#[allow(unused)] pub fn default_static_addrs() -> HashMap> { let mut m = HashMap::new(); m.insert(NYM_API_DOMAIN.to_string(), NYM_API_IPS.to_vec()); diff --git a/common/http-api-client/src/dns/static_resolver.rs b/common/http-api-client/src/dns/static_resolver.rs index 8660044735..7c2dbed576 100644 --- a/common/http-api-client/src/dns/static_resolver.rs +++ b/common/http-api-client/src/dns/static_resolver.rs @@ -14,42 +14,78 @@ const DEFAULT_PRE_RESOLVE_TIMEOUT: Duration = super::DEFAULT_POSITIVE_LOOKUP_CAC #[derive(Debug, Default, Clone)] pub struct StaticResolver { - static_addr_map: Arc>>, + fallback_addr_map: Arc>>>, + preresolve_addr_map: Arc>>, pre_resolve_timeout: Option, } +#[derive(Debug, Clone, Default)] +enum PreResolveStatus { + #[default] + Valid, + ValidUntil(Instant), +} + #[derive(Debug, Clone, Default)] struct Entry { - valid_for_pre_resolve_until: Option, + status: PreResolveStatus, addrs: Vec, } impl Entry { fn new(addrs: Vec) -> Self { Self { - valid_for_pre_resolve_until: None, + status: PreResolveStatus::Valid, addrs, } } + + fn new_timeout(addrs: Vec, timeout: Duration) -> Self { + Self { + status: PreResolveStatus::ValidUntil(Instant::now() + timeout), + addrs, + } + } + + fn is_valid(&self) -> bool { + match self.status { + PreResolveStatus::Valid => true, + PreResolveStatus::ValidUntil(t) => t > Instant::now(), + } + } } impl StaticResolver { - pub fn new(static_entries: HashMap>) -> StaticResolver { - debug!("building static resolver"); - let static_entries = static_entries - .into_iter() - .map(|(name, ips)| (name, Entry::new(ips))) - .collect(); + pub fn new() -> StaticResolver { Self { - static_addr_map: Arc::new(Mutex::new(static_entries)), + fallback_addr_map: Arc::new(Mutex::new(HashMap::new())), + preresolve_addr_map: Arc::new(Mutex::new(HashMap::new())), pre_resolve_timeout: Some(DEFAULT_PRE_RESOLVE_TIMEOUT), } } - /// Return the full set of domain names and associated addresses stored in this static lookup table - pub fn get_addrs(&self) -> HashMap> { + /// Initialize the contents of the pre-resolve table for this instance of the static resolver + #[allow(unused)] + pub fn with_preresolve(mut self, entries: HashMap>) -> Self { + let entries = entries + .into_iter() + .map(|(name, ips)| (name, Entry::new(ips))) + .collect(); + self.preresolve_addr_map = Arc::new(Mutex::new(entries)); + self + } + + /// Initialize the contenes of the fallback table for this instance of the static resolver + pub fn with_fallback(mut self, entries: HashMap>) -> Self { + self.fallback_addr_map = Arc::new(Mutex::new(entries)); + self + } + + /// Return the set of domain names and associated addresses stored in the pre-resolve static + /// lookup table + pub fn get_preresolve_addrs(&self) -> HashMap> { let mut out = HashMap::new(); - self.static_addr_map + self.preresolve_addr_map .lock() .unwrap() .iter() @@ -59,6 +95,38 @@ impl StaticResolver { out } + /// Return the set of domain names and associated addresses stored in the fallback static lookup + /// table + pub fn get_fallback_addrs(&self) -> HashMap> { + self.fallback_addr_map.lock().unwrap().clone() + } + + /// Set (or overwrite) the map of static addresses to be returned only after attempting a lookup + /// over the network resolver. + pub fn set_fallback(&self, addrs: HashMap>) { + self.fallback_addr_map.lock().unwrap().extend(addrs); + } + + /// Clear entries from the static table that would return entries during the pre-resolve stage. + /// This means that all lookups will attempt to use the network resolver again before the static + /// table is consulted. + /// + /// Entries elevated to pre-resolve from fallback (added from default or using + /// [`set_fallback`]`) will have their cache timeout cleared. Entries added directly to + /// pre-resolve (using [`Self::preresolve_to_addrs`]) will be removed. + pub fn clear_preresolve(&self) { + *self.preresolve_addr_map.lock().unwrap() = HashMap::new(); + } + + /// Set (or overwrite) the map of static addresses and mark these domains to be returned + /// WITHOUT attempting a lookup over the network resolver. + pub fn set_preresolve(&self, addrs: HashMap>) { + let mut current_map = self.preresolve_addr_map.lock().unwrap(); + for (domain, ips) in addrs.into_iter() { + _ = current_map.insert(domain, Entry::new(ips)) + } + } + /// Change the timeout for which domains can be pre-resolved after they are looked up in the /// static lookup table. #[allow(unused)] @@ -71,44 +139,58 @@ impl StaticResolver { /// recently (within the configured timeout) looked it up previously in this static table using /// a regular resolve. pub fn pre_resolve(&self, name: &str) -> Option> { - debug!("found {name:?} in pre-resolve static table resolver"); - - self.pre_resolve_timeout?; - - self.static_addr_map + self.preresolve_addr_map .lock() .unwrap() .get(name) - .filter(|e| { - e.valid_for_pre_resolve_until - .is_some_and(|t| t > Instant::now()) + .filter(|entry| entry.is_valid()) + .map(|entry| { + debug!("pre-resolve lookup hit for \"{name:?}\" in static table resolver"); + entry.addrs.clone() }) - .map(|e| e.addrs.clone()) } #[allow(unused)] pub fn resolve_str(&self, name: &str) -> Option> { Self::resolve_inner( - self.static_addr_map.lock().unwrap(), + self.fallback_addr_map.lock().unwrap(), + self.preresolve_addr_map.lock().unwrap(), name, self.pre_resolve_timeout, ) - .map(|e| e.addrs) } fn resolve_inner( - mut table: MutexGuard<'_, HashMap>, + fallback_table: MutexGuard<'_, HashMap>>, + mut preresolve_table: MutexGuard<'_, HashMap>, name: &str, - timeout: Option, - ) -> Option { - let resolved = table.get_mut(name)?; + pre_resolve_cache_timeout: Option, + ) -> Option> { + let resolved = fallback_table.get(name)?; - debug!("found {name:?} in static table resolver"); + debug!("lookup hit for \"{name:?}\" in static table resolver"); - if let Some(pre_resolve_timeout) = timeout { - // We had to look this entry up and a pre-resolve duration is defined, so it will - // trigger in pre-resolve lookups for the next _timeout_ window. - resolved.valid_for_pre_resolve_until = Some(Instant::now() + pre_resolve_timeout); + // We had to look this entry up and a pre-resolve duration is defined, so it will + // trigger in pre-resolve lookups for the next _timeout_ window if it wasn't already + // triggering. + if let Some(pre_resolve_timeout) = pre_resolve_cache_timeout { + match preresolve_table.get_mut(name) { + None => { + _ = preresolve_table.insert( + name.to_string(), + Entry::new_timeout(resolved.clone(), pre_resolve_timeout), + ); + } + // Not sure how we would get cases where this is Some( ) -- it requires having a + // Valid entry in the preresolve table and still doing a lookup against fallback. + Some(entry) if matches!(entry.status, PreResolveStatus::ValidUntil(_)) => { + _ = preresolve_table.insert( + name.to_string(), + Entry::new_timeout(resolved.clone(), pre_resolve_timeout), + ); + } + _ => {} + } } Some(resolved.clone()) } @@ -117,13 +199,23 @@ impl StaticResolver { impl Resolve for StaticResolver { fn resolve(&self, name: Name) -> Resolving { debug!("looking up {name:?} in static resolver"); - let addr_map = self.static_addr_map.clone(); + // these should clone arcs, not the actual tables + let fallback_addr_map = self.fallback_addr_map.clone(); + let presesolve_addr_map = self.preresolve_addr_map.clone(); let timeout = self.pre_resolve_timeout; + // Also the returned future doesn't try to take the lock on the tables until the + // future is awaited, so no blocking issues. Box::pin(async move { - let addr_map = addr_map.lock().unwrap(); - let lookup = match Self::resolve_inner(addr_map, name.as_str(), timeout) { + let fallback_addr_map = fallback_addr_map.lock().unwrap(); + let presesolve_addr_map = presesolve_addr_map.lock().unwrap(); + let lookup = match Self::resolve_inner( + fallback_addr_map, + presesolve_addr_map, + name.as_str(), + timeout, + ) { None => return Err(ResolveError::StaticLookupMiss.into()), - Some(entry) => entry.addrs, + Some(addrs) => addrs, }; let addrs: Addrs = Box::new( lookup @@ -142,6 +234,7 @@ mod test { use super::*; use std::error::Error as StdError; + use std::net::Ipv4Addr; use std::str::FromStr; #[tokio::test] @@ -149,7 +242,7 @@ mod test { let example_domain = String::from("static.nymvpn.com"); // lookup for domain for which there is no entry - let resolver = StaticResolver::new(HashMap::new()); + let resolver = StaticResolver::new(); let url = reqwest::dns::Name::from_str(&example_domain).unwrap(); let result = resolver.resolve(url).await; @@ -166,7 +259,7 @@ mod test { addr_map.insert(example_domain.clone(), vec![example_ip4, example_ip6]); let url = reqwest::dns::Name::from_str(&example_domain).unwrap(); - let resolver = StaticResolver::new(addr_map); + let resolver = StaticResolver::new().with_fallback(addr_map); let mut addrs = resolver.resolve(url).await?; assert!(addrs.contains(&SocketAddr::new(example_ip4, 0))); assert!(addrs.contains(&SocketAddr::new(example_ip6, 0))); @@ -175,7 +268,7 @@ mod test { } #[test] - fn static_lookup_pre_resolve() { + fn elevate_fallback_to_pre_resolve() { let example_duration = Duration::from_secs(3); let example_domain = String::from("static.nymvpn.com"); let mut addr_map = HashMap::new(); @@ -183,24 +276,23 @@ mod test { let example_ip6: IpAddr = "dead::beef".parse().unwrap(); addr_map.insert(example_domain.clone(), vec![example_ip4, example_ip6]); - let resolver = StaticResolver::new(addr_map).with_pre_resolve_timeout(example_duration); + let resolver = StaticResolver::new() + .with_fallback(addr_map) + .with_pre_resolve_timeout(example_duration); // ensure that attempting to pre-resolve without first resolving returns none let result = resolver.pre_resolve(&example_domain); assert!(result.is_none()); // resolving should now update the pre-resolve validity timeout for the entry - let entry = StaticResolver::resolve_inner( - resolver.static_addr_map.lock().unwrap(), - &example_domain, - Some(example_duration), - ) - .expect("missing entry???!!!!"); - assert!( - entry - .valid_for_pre_resolve_until - .is_some_and(|t| t < Instant::now() + example_duration) - ); + let _addrs = resolver + .resolve_str(&example_domain) + .expect("entry should exist"); + assert!(matches!( + resolver.preresolve_status(&example_domain), + Some(PreResolveStatus::ValidUntil(t)) + if t < Instant::now() + example_duration + )); // check that pre-resolve now returns the expected record let addrs = resolver @@ -214,4 +306,139 @@ mod test { let result = resolver.pre_resolve(&example_domain); assert!(result.is_none()); } + + #[test] + fn set_and_use_preresolve() { + let example_duration = Duration::from_secs(3); + let example_domains = [ + String::from("static1.nymvpn.com"), + String::from("static2.nymvpn.com"), + String::from("preresolve.nymvpn.com"), + ]; + let mut addr_map1 = HashMap::new(); + addr_map1.insert( + example_domains[0].clone(), + vec![Ipv4Addr::new(10, 10, 10, 10).into()], + ); + addr_map1.insert( + example_domains[1].clone(), + vec![Ipv4Addr::new(1, 1, 1, 1).into()], + ); + + let mut addr_map2 = HashMap::new(); + addr_map2.insert( + example_domains[1].clone(), + vec![Ipv4Addr::new(1, 1, 1, 1).into()], + ); + addr_map2.insert( + example_domains[2].clone(), + vec![Ipv4Addr::new(8, 8, 8, 8).into()], + ); + + let resolver = StaticResolver::new() + .with_fallback(addr_map1) + .with_pre_resolve_timeout(example_duration); + + // Attempting to pre-resolve without setting the table returns none + let result = resolver.pre_resolve(&example_domains[0]); + assert!(result.is_none()); + + resolver.set_preresolve(addr_map2); + + // After setting the pre-resolve, addresses in the the table are returned + let result = resolver.pre_resolve(&example_domains[1]); + assert!(result.is_some()); + + // If the domain wasn't in the pre-resolve table it returns none. + let result = resolver.pre_resolve(&example_domains[0]); + assert!(result.is_none()); + + resolver.clear_preresolve(); + } + + #[test] + fn preresolve_with_fallback() { + let example_duration = Duration::from_secs(3); + let example_domains = [ + String::from("static1.nymvpn.com"), + String::from("static2.nymvpn.com"), + String::from("preresolve.nymvpn.com"), + ]; + let mut addr_map1 = HashMap::new(); + addr_map1.insert( + example_domains[0].clone(), + vec![Ipv4Addr::new(10, 10, 10, 10).into()], + ); + addr_map1.insert( + example_domains[1].clone(), + vec![Ipv4Addr::new(1, 1, 1, 1).into()], + ); + + let mut addr_map2 = HashMap::new(); + addr_map2.insert( + example_domains[1].clone(), + vec![Ipv4Addr::new(1, 1, 1, 1).into()], + ); + addr_map2.insert( + example_domains[2].clone(), + vec![Ipv4Addr::new(8, 8, 8, 8).into()], + ); + + let resolver = StaticResolver::new() + .with_fallback(addr_map1) + .with_preresolve(addr_map2) + .with_pre_resolve_timeout(example_duration); + + // when using both pre-resolve and fallback elevating entries from fallback to pre-resolve + // leaves the entries as `Valid`. + assert!(matches!( + resolver.preresolve_status(&example_domains[1]), + Some(PreResolveStatus::Valid) + )); + let _addrs = resolver + .resolve_str(&example_domains[1]) + .expect("entry should exist"); + assert!(matches!( + resolver.preresolve_status(&example_domains[1]), + Some(PreResolveStatus::Valid) + )); + + // entries not already in pre-resolve get elevated with a timeout. + assert!(!resolver.preresolve_contains(&example_domains[0])); + let _addrs = resolver + .resolve_str(&example_domains[0]) + .expect("entry should exist"); + assert!(resolver.preresolve_contains(&example_domains[0])); + assert!(matches!( + resolver.preresolve_status(&example_domains[0]), + Some(PreResolveStatus::ValidUntil(_)) + )); + + // clearing the pre-resolve table doesn't impact the fallback table. + resolver.clear_preresolve(); + assert!(!resolver.preresolve_contains(&example_domains[0])); + assert!(!resolver.preresolve_contains(&example_domains[1])); + assert!(!resolver.preresolve_contains(&example_domains[2])); + assert!(!resolver.fallback_contains(&example_domains[0])); + assert!(!resolver.fallback_contains(&example_domains[1])); + } + + /// convenience functions for testing + impl StaticResolver { + fn preresolve_status(&self, name: &str) -> Option { + self.preresolve_addr_map + .lock() + .unwrap() + .get(name) + .map(|e| e.status.clone()) + } + + fn preresolve_contains(&self, name: &str) -> bool { + self.preresolve_addr_map.lock().unwrap().contains_key(name) + } + + fn fallback_contains(&self, name: &str) -> bool { + self.preresolve_addr_map.lock().unwrap().contains_key(name) + } + } } diff --git a/common/http-api-client/src/fronted.rs b/common/http-api-client/src/fronted.rs index be884264cf..e7da0a36c0 100644 --- a/common/http-api-client/src/fronted.rs +++ b/common/http-api-client/src/fronted.rs @@ -3,15 +3,21 @@ //! Utilities for and implementation of request tunneling -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{ + Arc, LazyLock, RwLock, + atomic::{AtomicBool, Ordering}, +}; use tracing::warn; -use crate::ClientBuilder; +use crate::{Client, ClientBuilder}; + +static SHARED_FRONTING_POLICY: LazyLock>> = + LazyLock::new(|| Arc::new(RwLock::new(FrontPolicy::Off))); // #[cfg(feature = "tunneling")] #[derive(Debug)] pub(crate) struct Front { - pub(crate) policy: FrontPolicy, + pub(crate) policy: Arc>, enabled: AtomicBool, } @@ -19,7 +25,7 @@ impl Clone for Front { fn clone(&self) -> Self { Self { policy: self.policy.clone(), - enabled: AtomicBool::new(self.enabled.load(Ordering::Relaxed)), + enabled: AtomicBool::new(false), } } } @@ -27,13 +33,30 @@ impl Clone for Front { impl Front { pub(crate) fn new(policy: FrontPolicy) -> Self { Self { - enabled: AtomicBool::new(policy == FrontPolicy::Always), + enabled: AtomicBool::new(false), + policy: Arc::new(RwLock::new(policy)), + } + } + + pub(crate) fn off() -> Self { + Self::new(FrontPolicy::Off) + } + + pub(crate) fn shared() -> Self { + let policy = SHARED_FRONTING_POLICY.clone(); + Self { + enabled: AtomicBool::new(false), policy, } } + pub(crate) fn set_policy(&self, policy: FrontPolicy) { + *self.policy.write().unwrap() = policy; + self.enabled.store(false, Ordering::Relaxed); + } + pub(crate) fn is_enabled(&self) -> bool { - match self.policy { + match *self.policy.read().unwrap() { FrontPolicy::Off => false, FrontPolicy::OnRetry => self.enabled.load(Ordering::Relaxed), FrontPolicy::Always => true, @@ -46,14 +69,13 @@ impl Front { if self.is_enabled() { return; } - if matches!(self.policy, FrontPolicy::OnRetry) { + if matches!(*self.policy.read().unwrap(), FrontPolicy::OnRetry) { self.enabled.store(true, Ordering::Relaxed); } } } #[derive(Debug, Default, PartialEq, Clone)] -#[cfg(feature = "tunneling")] /// Policy for when to use domain fronting for HTTP requests. pub enum FrontPolicy { /// Always use domain fronting for all requests. @@ -66,29 +88,208 @@ pub enum FrontPolicy { } impl ClientBuilder { - /// Enable and configure request tunneling for API requests. - #[cfg(feature = "tunneling")] - pub fn with_fronting(mut self, policy: FrontPolicy) -> Self { - let front = Front::new(policy); + /// Enable and configure request tunneling for API requests. If no front policy is + /// provided the shared fronting policy will be used. + pub fn with_fronting(mut self, policy: Option) -> Self { + let front = if let Some(p) = policy { + Front::new(p) + } else { + Front::shared() + }; // Check if any of the supplied urls even support fronting if !self.urls.iter().any(|url| url.has_front()) { warn!( - "fronting is enabled, but none of the supplied urls have configured fronting domains" + "fronting is enabled, but none of the supplied urls have configured fronting domains: {:?}", + self.urls ); } - self.front = Some(front); + self.front = front; self } } +impl Client { + /// Set the policy for enabling fronting. If fronting was previously unset this will set it, and + /// make it possible to enable (i.e [`FrontPolicy::Off`] will not enable it). + /// + /// Calling this function sets a custom policy for this client, disconnecting it from the shared + /// fronting policy -- i.e. changes applied through [`Client::set_shared_front_policy`] will not + /// be impact this client. + pub fn set_front_policy(&mut self, policy: FrontPolicy) { + self.front.set_policy(policy) + } + + /// Set the fronting policy for this client to follow the shared policy. + pub fn use_shared_front_policy(&mut self) { + self.front = Front::shared(); + } + + /// Set the fronting policy for all clients using the shared policy. + // + // NOTE: this does not reset the per-instance enabled flag like it will when using + // [`Front::set_front_policy`]. So if a client is using shared policy with the `OnRetry` policy + // and this function is used to swap that policy away from and then back to `OnRetry` the + // fronting will still be enabled. Noting this here just in case this triggers any corner cases + // down the road. + pub fn set_shared_front_policy(policy: FrontPolicy) { + *SHARED_FRONTING_POLICY.write().unwrap() = policy; + } +} + #[cfg(test)] mod tests { use super::*; use crate::{ApiClientCore, NO_PARAMS, Url}; + impl Front { + pub(crate) fn policy(&self) -> FrontPolicy { + self.policy.read().unwrap().clone() + } + } + + /// Policy can be set for an independent client and the update is applied properly + #[test] + fn set_policy_independent_client() { + let url1 = Url::new( + "https://validator.global.ssl.fastly.net", + Some(vec!["https://yelp.global.ssl.fastly.net"]), + ) + .unwrap(); + + let mut client1 = ClientBuilder::new(url1.clone()) + .unwrap() + .with_fronting(Some(FrontPolicy::Off)) + .build() + .unwrap(); + assert!(client1.front.policy() == FrontPolicy::Off); + + let client2 = ClientBuilder::new(url1.clone()) + .unwrap() + .with_fronting(Some(FrontPolicy::OnRetry)) + .build() + .unwrap(); + + // Ensure that setting the policy for a client it gets properly applied. + client1.set_front_policy(FrontPolicy::Always); + assert!(client1.front.policy() == FrontPolicy::Always); + + // ensure that setting the policy in a client NOT using the shared policy does NOT update + // the policy used by another client. + assert!(client2.front.policy() == FrontPolicy::OnRetry); + + // Ensure that the policy takes effect and is applied when setting host headers on outgoing + // requests + let req = client1 + .create_request(reqwest::Method::GET, &["/"], NO_PARAMS, None::<&()>) + .unwrap() + .build() + .unwrap(); + + let expected_host = url1.host_str().unwrap(); + assert!( + req.headers() + .get(reqwest::header::HOST) + .is_some_and(|h| h.to_str().unwrap() == expected_host), + "{:?} != {:?}", + expected_host, + req, + ); + + let expected_front = url1.front_str().unwrap(); + assert!( + req.url() + .host() + .is_some_and(|url| url.to_string() == expected_front), + "{:?} != {:?}", + expected_front, + req, + ); + } + + /// Policy can be set for the shared client and the update is applied properly + // NOTE THIS TEST IS DISABLED BECAUSE IT INTERACTS WITH THE SHARED POLICY AND AS SUCH CAN HAVE + // AN IMPACT ON OTHER TESTS + #[test] + #[cfg(any())] // #[ignore] we run --ignore in CI/CD assuming it just means slow -_- + fn set_policy_shared_client() { + let url1 = Url::new( + "https://validator.global.ssl.fastly.net", + Some(vec!["https://yelp.global.ssl.fastly.net"]), + ) + .unwrap(); + + Client::set_shared_front_policy(FrontPolicy::Off); + assert!(*SHARED_FRONTING_POLICY.read().unwrap() == FrontPolicy::Off); + + let client1 = ClientBuilder::new(url1.clone()) + .unwrap() + .with_fronting(None) + .build() + .unwrap(); + assert!(client1.front.policy() == FrontPolicy::Off); + + let mut client2 = ClientBuilder::new(url1.clone()) + .unwrap() + .with_fronting(Some(FrontPolicy::Off)) + .build() + .unwrap(); + + // Ensure that setting the shared policy gets properly applied + Client::set_shared_front_policy(FrontPolicy::Always); + assert!(client1.front.policy() == FrontPolicy::Always); + + // Setting the shared policy should NOT update clients NOT using the shared policy. + assert!(client2.front.policy() == FrontPolicy::Off); + + // Ensure that the policy takes effect and is applied when setting host headers on outgoing + // requests + let req = client1 + .create_request(reqwest::Method::GET, &["/"], NO_PARAMS, None::<&()>) + .unwrap() + .build() + .unwrap(); + + let expected_host = url1.host_str().unwrap(); + assert!( + req.headers() + .get(reqwest::header::HOST) + .is_some_and(|h| h.to_str().unwrap() == expected_host), + "{:?} != {:?}", + expected_host, + req, + ); + + let expected_front = url1.front_str().unwrap(); + assert!( + req.url() + .host() + .is_some_and(|url| url.to_string() == expected_front), + "{:?} != {:?}", + expected_front, + req, + ); + + // ensure that setting to the shared policy works + client2.use_shared_front_policy(); + assert!(client2.front.policy() == FrontPolicy::Always); + + // ensure that if the policy is OnRetry then the `enabled` fields are still independent, + // despite the policy being shared. + Client::set_shared_front_policy(FrontPolicy::OnRetry); + assert!(client1.front.policy() == FrontPolicy::OnRetry); + assert!(client2.front.policy() == FrontPolicy::OnRetry); + + assert!(!client1.front.is_enabled()); + assert!(!client2.front.is_enabled()); + + client1.front.retry_enable(); + assert!(client1.front.is_enabled()); + assert!(!client2.front.is_enabled()); + } + #[tokio::test] async fn nym_api_works() { let url1 = Url::new( @@ -104,7 +305,7 @@ mod tests { let client = ClientBuilder::new(url1) .expect("bad url") - .with_fronting(FrontPolicy::Always) + .with_fronting(Some(FrontPolicy::Always)) .build() .expect("failed to build client"); @@ -140,7 +341,7 @@ mod tests { let client = ClientBuilder::new_with_urls(vec![url1, url2]) .expect("bad url") - .with_fronting(FrontPolicy::Always) + .with_fronting(Some(FrontPolicy::Always)) .build() .expect("failed to build client"); diff --git a/common/http-api-client/src/lib.rs b/common/http-api-client/src/lib.rs index 06a0511564..20b2ffcfd7 100644 --- a/common/http-api-client/src/lib.rs +++ b/common/http-api-client/src/lib.rs @@ -136,6 +136,7 @@ //! ``` #![warn(missing_docs)] +use http::header::USER_AGENT; pub use inventory; pub use reqwest; pub use reqwest::ClientBuilder as ReqwestClientBuilder; @@ -147,6 +148,7 @@ pub mod registry; use crate::path::RequestPath; use async_trait::async_trait; use bytes::Bytes; +use cfg_if::cfg_if; use http::HeaderMap; use http::header::{ACCEPT, CONTENT_TYPE}; use itertools::Itertools; @@ -161,9 +163,7 @@ use std::time::Duration; use thiserror::Error; use tracing::{debug, instrument, warn}; -#[cfg(not(target_arch = "wasm32"))] -use std::net::SocketAddr; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; #[cfg(feature = "tunneling")] mod fronted; @@ -195,6 +195,8 @@ use nym_http_api_client_macro::client_defaults; /// high and chatty protocols take a while to complete. pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); +const NYM_OUTER_SNI_HEADER: &str = "NYM-ORIGINAL-OUTER-SNI"; + #[cfg(not(target_arch = "wasm32"))] client_defaults!( priority = -100; @@ -206,6 +208,24 @@ client_defaults!( user_agent = format!("nym-http-api-client/{}", env!("CARGO_PKG_VERSION")) ); +static SHARED_CLIENT: LazyLock = LazyLock::new(|| { + tracing::info!("Initializing shared HTTP client"); + cfg_if! { + if #[cfg(target_arch = "wasm32")] { + reqwest::ClientBuilder::new().build() + .expect("failed to initialize shared http client") + } else { + let mut builder = default_builder(); + + builder = builder.dns_resolver(Arc::new(HickoryDnsResolver::default())); + + builder + .build() + .expect("failed to initialize shared http client") + } + } +}); + /// Collection of URL Path Segments pub type PathSegments<'a> = &'a [&'a str]; /// Collection of HTTP Request Parameters @@ -327,6 +347,12 @@ pub enum HttpClientError { source: reqwest::Error, }, + #[error("failed to parse header value: {source}")] + InvalidHeaderValue { + #[source] + source: http::Error, + }, + #[error("failed to send request for {url}: {source}")] RequestSendFailure { url: Box, @@ -554,6 +580,19 @@ pub trait ApiClientCore { let req = self.create_request(method, path, params, json_body)?; self.send(req).await } + + /// If multiple base urls are available rotate to next (e.g. when the current one resulted in an error) + /// + /// Takes an optional URL argument. If this is none, the current host will be updated automatically. + /// If a url is provided first check that the CURRENT host matches the hostname in the URL before + /// triggering a rotation. This is meant to prevent parallel requests that fail from rotating the host + /// multiple times. + fn maybe_rotate_hosts(&self, offending_url: Option); + + /// If the fronting policy for the client is set to `OnRetry` this function will enable the + /// fronting if not already enabled. + #[cfg(feature = "tunneling")] + fn maybe_enable_fronting(&self, context: impl std::fmt::Debug); } /// A `ClientBuilder` can be used to create a [`Client`] with custom configuration applied consistently @@ -562,16 +601,18 @@ pub struct ClientBuilder { urls: Vec, timeout: Option, - custom_user_agent: bool, - reqwest_client_builder: reqwest::ClientBuilder, + custom_user_agent: Option, + reqwest_client_builder: Option, #[allow(dead_code)] // not dead code, just unused in wasm use_secure_dns: bool, #[cfg(feature = "tunneling")] - front: Option, + front: fronted::Front, retry_limit: usize, serialization: SerializationFormat, + + error: Option, } impl ClientBuilder { @@ -642,10 +683,10 @@ impl ClientBuilder { let mut builder = Self::new_with_urls(urls)?; - // Enable domain fronting by default (on retry) + // Enable domain fronting using the shared fronting policy #[cfg(feature = "tunneling")] { - builder = builder.with_fronting(FrontPolicy::OnRetry); + builder = builder.with_fronting(None); } Ok(builder) @@ -659,26 +700,31 @@ impl ClientBuilder { let urls = Self::check_urls(urls); - #[cfg(target_arch = "wasm32")] - let reqwest_client_builder = reqwest::ClientBuilder::new(); - - #[cfg(not(target_arch = "wasm32"))] - let reqwest_client_builder = default_builder(); - Ok(ClientBuilder { urls, timeout: None, - custom_user_agent: false, - reqwest_client_builder, + custom_user_agent: None, + reqwest_client_builder: None, use_secure_dns: true, #[cfg(feature = "tunneling")] - front: None, + front: fronted::Front::off(), retry_limit: 0, serialization: SerializationFormat::Json, + error: None, }) } + /// Configure use of an independent HTTP request executor. This prevents use of beneficial + /// features like connection pooling under the hood. + #[cfg(not(target_arch = "wasm32"))] + pub fn non_shared(mut self) -> Self { + if self.reqwest_client_builder.is_none() { + self.reqwest_client_builder = Some(default_builder()); + } + self + } + /// Add an additional URL to the set usable by this constructed `Client` pub fn add_url(mut self, url: Url) -> Self { self.urls.push(url); @@ -723,7 +769,7 @@ impl ClientBuilder { /// Provide a pre-configured [`reqwest::ClientBuilder`] pub fn with_reqwest_builder(mut self, reqwest_builder: reqwest::ClientBuilder) -> Self { - self.reqwest_client_builder = reqwest_builder; + self.reqwest_client_builder = Some(reqwest_builder); self } @@ -733,18 +779,12 @@ impl ClientBuilder { V: TryInto, V::Error: Into, { - self.custom_user_agent = true; - self.reqwest_client_builder = self.reqwest_client_builder.user_agent(value); - self - } - - /// Override DNS resolution for specific domains to particular IP addresses. - /// - /// Set the port to `0` to use the conventional port for the given scheme (e.g. 80 for http). - /// Ports in the URL itself will always be used instead of the port in the overridden addr. - #[cfg(not(target_arch = "wasm32"))] - pub fn resolve_to_addrs(mut self, domain: &str, addrs: &[SocketAddr]) -> ClientBuilder { - self.reqwest_client_builder = self.reqwest_client_builder.resolve_to_addrs(domain, addrs); + match value.try_into() { + Ok(v) => self.custom_user_agent = Some(v), + Err(err) => { + self.error = Some(HttpClientError::InvalidHeaderValue { source: err.into() }) + } + } self } @@ -761,30 +801,33 @@ impl ClientBuilder { /// Returns a Client that uses this ClientBuilder configuration. pub fn build(self) -> Result { + if let Some(err) = self.error { + return Err(err); + } + #[cfg(target_arch = "wasm32")] - let reqwest_client = self.reqwest_client_builder.build()?; + let reqwest_client = Some(reqwest::ClientBuilder::new().build()?); - // TODO: we should probably be propagating the error rather than panicking, - // but that'd break bunch of things due to type changes #[cfg(not(target_arch = "wasm32"))] - let reqwest_client = { - let mut builder = self.reqwest_client_builder; + let reqwest_client = self + .reqwest_client_builder + .map(|mut builder| { + // unless explicitly disabled use the DoT/DoH enabled resolver + if self.use_secure_dns { + builder = builder.dns_resolver(Arc::new(HickoryDnsResolver::default())); + } - // unless explicitly disabled use the DoT/DoH enabled resolver - if self.use_secure_dns { - builder = builder.dns_resolver(Arc::new(HickoryDnsResolver::default())); - } - - builder - .build() - .map_err(HttpClientError::reqwest_client_build_error)? - }; + builder + .build() + .map_err(HttpClientError::reqwest_client_build_error) + }) + .transpose()?; let client = Client { base_urls: self.urls, current_idx: Arc::new(AtomicUsize::new(0)), reqwest_client, - using_secure_dns: self.use_secure_dns, + custom_user_agent: self.custom_user_agent, #[cfg(feature = "tunneling")] front: self.front, @@ -804,11 +847,11 @@ impl ClientBuilder { pub struct Client { base_urls: Vec, current_idx: Arc, - reqwest_client: reqwest::Client, - using_secure_dns: bool, + reqwest_client: Option, + custom_user_agent: Option, #[cfg(feature = "tunneling")] - front: Option, + front: fronted::Front, #[cfg(target_arch = "wasm32")] request_timeout: Duration, @@ -862,8 +905,8 @@ impl Client { Client { base_urls: vec![new_url], current_idx: Arc::new(Default::default()), - reqwest_client: self.reqwest_client.clone(), - using_secure_dns: self.using_secure_dns, + reqwest_client: None, + custom_user_agent: None, #[cfg(feature = "tunneling")] front: self.front.clone(), @@ -897,9 +940,7 @@ impl Client { #[cfg(feature = "tunneling")] fn matches_current_host(&self, url: &Url) -> bool { - if let Some(ref front) = self.front - && front.is_enabled() - { + if self.front.is_enabled() { url.host_str() == self.current_url().front_str() } else { url.host_str() == self.current_url().host_str() @@ -926,9 +967,7 @@ impl Client { } #[cfg(feature = "tunneling")] - if let Some(ref front) = self.front - && front.is_enabled() - { + if self.front.is_enabled() { // if we are using fronting, try updating to the next front let url = self.current_url(); @@ -948,9 +987,7 @@ impl Client { // if fronting is enabled we want to update to a host that has fronts configured #[cfg(feature = "tunneling")] - if let Some(ref front) = self.front - && front.is_enabled() - { + if self.front.is_enabled() { while next != orig { if self.base_urls[next].has_front() { // we have a front for the next host, so we can use it @@ -981,14 +1018,12 @@ impl Client { /// this method. For example, if the client is configured to rotate hosts after each error, this /// method should be called after the host has been updated -- i.e. as part of the subsequent /// send. - fn apply_hosts_to_req(&self, r: &mut reqwest::Request) -> (&str, Option<&str>) { + pub(crate) fn apply_hosts_to_req(&self, r: &mut reqwest::Request) -> (&str, Option<&str>) { let url = self.current_url(); r.url_mut().set_host(url.host_str()).unwrap(); #[cfg(feature = "tunneling")] - if let Some(ref front) = self.front - && front.is_enabled() - { + if self.front.is_enabled() { if let Some(front_host) = url.front_str() { if let Some(actual_host) = url.host_str() { tracing::debug!( @@ -1008,6 +1043,13 @@ impl Client { .headers_mut() .insert(reqwest::header::HOST, actual_host_header); + // Set a custom header to capture the outer host (used in the SNI) of the request + let front_host_header: HeaderValue = + front_host.parse().unwrap_or(HeaderValue::from_static("")); + _ = r + .headers_mut() + .insert(NYM_OUTER_SNI_HEADER, front_host_header); + return (url.as_str(), url.front_str()); } else { tracing::debug!( @@ -1048,12 +1090,21 @@ impl ApiClientCore for Client { self.apply_hosts_to_req(&mut req); - let mut rb = RequestBuilder::from_parts(self.reqwest_client.clone(), req); + let client = if let Some(client) = &self.reqwest_client { + client.clone() + } else { + SHARED_CLIENT.clone() + }; + let mut rb = RequestBuilder::from_parts(client, req); rb = rb .header(ACCEPT, self.serialization.content_type()) .header(CONTENT_TYPE, self.serialization.content_type()); + if let Some(user_agent) = &self.custom_user_agent { + rb = rb.header(USER_AGENT, user_agent.clone()); + } + if let Some(body) = body { match self.serialization { SerializationFormat::Json => { @@ -1096,16 +1147,19 @@ impl ApiClientCore for Client { #[cfg(target_arch = "wasm32")] let response: Result = { - Ok(wasmtimer::tokio::timeout( - self.request_timeout, - self.reqwest_client.execute(req), + let client = self.reqwest_client.as_ref().unwrap_or(&*SHARED_CLIENT); + Ok( + wasmtimer::tokio::timeout(self.request_timeout, client.execute(req)) + .await + .map_err(|_timeout| HttpClientError::RequestTimeout)??, ) - .await - .map_err(|_timeout| HttpClientError::RequestTimeout)??) }; #[cfg(not(target_arch = "wasm32"))] - let response = self.reqwest_client.execute(req).await; + let response = { + let client = self.reqwest_client.as_ref().unwrap_or(&*SHARED_CLIENT); + client.execute(req).await + }; match response { Ok(resp) => return Ok(resp), @@ -1121,20 +1175,10 @@ impl ApiClientCore for Client { if is_network_err { // if we have multiple urls, update to the next - self.update_host(Some(url.clone())); + self.maybe_rotate_hosts(Some(url.clone())); #[cfg(feature = "tunneling")] - if let Some(ref front) = self.front { - // If fronting is set to be enabled on error, enable domain fronting as we - // have encountered an error. - let was_enabled = front.is_enabled(); - front.retry_enable(); - if !was_enabled && front.is_enabled() { - tracing::info!( - "Domain fronting activated after connection failure: {err}", - ); - } - } + self.maybe_enable_fronting(("network", url.as_str(), &err)); } if attempts < self.retry_limit { @@ -1158,6 +1202,21 @@ impl ApiClientCore for Client { } } } + + fn maybe_rotate_hosts(&self, offending: Option) { + self.update_host(offending); + } + + #[cfg(feature = "tunneling")] + fn maybe_enable_fronting(&self, context: impl std::fmt::Debug) { + // If fronting is set to be OnRetry, enable domain fronting as we + // have encountered an error. + let was_enabled = self.front.is_enabled(); + self.front.retry_enable(); + if !was_enabled && self.front.is_enabled() { + tracing::debug!("Domain fronting activated after failure: {context:?}",); + } + } } /// Common usage functionality for the http client. @@ -1310,6 +1369,35 @@ pub trait ApiClient: ApiClientCore { self.get_response(path, params).await } + /// Attempt to parse a response object from an HTTP response + async fn parse_response( + &self, + res: Response, + allow_empty: bool, + ) -> Result + where + T: DeserializeOwned, + { + let url = Url::from(res.url()); + parse_response(res, allow_empty).await.inspect_err(|e| { + if matches!( + // if we encounter a read error while we attempt to parse it could be caused by censorship and we should + // rotate hosts / enable fronting. + e, + HttpClientError::ResponseReadFailure { + url: _, + headers: _, + status: _, + source: _, + } + ) { + self.maybe_rotate_hosts(Some(url.clone())); + #[cfg(feature = "tunneling")] + self.maybe_enable_fronting(("parse/read", url.as_str(), e)); + } + }) + } + /// 'get' data from the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple /// defined key-value parameters, e.g. `[("since", "12345")]`. Attempt to parse the response /// into the provided type `T` based on the content type header @@ -1327,7 +1415,8 @@ pub trait ApiClient: ApiClientCore { let res = self .send_request(reqwest::Method::GET, path, params, None::<&()>) .await?; - parse_response(res, false).await + + self.parse_response(res, false).await } /// 'post' json data to the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple @@ -1349,7 +1438,7 @@ pub trait ApiClient: ApiClientCore { let res = self .send_request(reqwest::Method::POST, path, params, Some(json_body)) .await?; - parse_response(res, false).await + self.parse_response(res, false).await } /// 'delete' json data from the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with @@ -1369,7 +1458,7 @@ pub trait ApiClient: ApiClientCore { let res = self .send_request(reqwest::Method::DELETE, path, params, None::<&()>) .await?; - parse_response(res, false).await + self.parse_response(res, false).await } /// 'patch' json data at the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple @@ -1391,7 +1480,7 @@ pub trait ApiClient: ApiClientCore { let res = self .send_request(reqwest::Method::PATCH, path, params, Some(json_body)) .await?; - parse_response(res, false).await + self.parse_response(res, false).await } /// `get` json data from the provided absolute endpoint, e.g. `"/api/v1/mixnodes?since=12345"`. @@ -1403,7 +1492,7 @@ pub trait ApiClient: ApiClientCore { { let req = self.create_request_endpoint(reqwest::Method::GET, endpoint, None::<&()>)?; let res = self.send(req).await?; - parse_response(res, false).await + self.parse_response(res, false).await } /// `post` json data to the provided absolute endpoint, e.g. `"/api/v1/mixnodes?since=12345"`. @@ -1420,7 +1509,7 @@ pub trait ApiClient: ApiClientCore { { let req = self.create_request_endpoint(reqwest::Method::POST, endpoint, Some(json_body))?; let res = self.send(req).await?; - parse_response(res, false).await + self.parse_response(res, false).await } /// `delete` json data from the provided absolute endpoint, e.g. @@ -1432,7 +1521,7 @@ pub trait ApiClient: ApiClientCore { { let req = self.create_request_endpoint(reqwest::Method::DELETE, endpoint, None::<&()>)?; let res = self.send(req).await?; - parse_response(res, false).await + self.parse_response(res, false).await } /// `patch` json data at the provided absolute endpoint, e.g. `"/api/v1/mixnodes?since=12345"`. @@ -1450,7 +1539,7 @@ pub trait ApiClient: ApiClientCore { let req = self.create_request_endpoint(reqwest::Method::PATCH, endpoint, Some(json_body))?; let res = self.send(req).await?; - parse_response(res, false).await + self.parse_response(res, false).await } } diff --git a/common/http-api-client/src/tests.rs b/common/http-api-client/src/tests.rs index b287f5fe78..9a72dfd72a 100644 --- a/common/http-api-client/src/tests.rs +++ b/common/http-api-client/src/tests.rs @@ -89,7 +89,8 @@ fn sanitizing_urls() { // - on error without retries is where we have multiple urls, is the url updated? #[tokio::test] -#[ignore] // test relies on external services being available and behaving in a specific way. +#[cfg(any())] // #[ignore] we run ignore assuming it just means slow in Ci/CD -_- +// test relies on external services being available and behaving in a specific way. async fn api_client_retry() -> Result<(), Box> { let client = ClientBuilder::new_with_urls(vec![ "http://broken.nym.test".parse()?, // This should fail because of DNS NXDomain (rotate) @@ -199,7 +200,7 @@ fn fronted_host_updating() { let url = Url::new("http://nym-api.test", Some(vec!["http://cdn1.test"])).unwrap(); let mut client = ClientBuilder::new(url) .unwrap() - .with_fronting(crate::fronted::FrontPolicy::Always) + .with_fronting(Some(crate::fronted::FrontPolicy::Always)) .build() .unwrap(); diff --git a/common/http-api-client/src/url.rs b/common/http-api-client/src/url.rs index 8b3b1a65b8..065df2558e 100644 --- a/common/http-api-client/src/url.rs +++ b/common/http-api-client/src/url.rs @@ -123,6 +123,16 @@ impl From for Url { } } +impl From<&reqwest::Url> for Url { + fn from(url: &url::Url) -> Self { + Self { + url: url.clone(), + fronts: None, + current_front: Arc::new(AtomicUsize::new(0)), + } + } +} + impl AsRef for Url { fn as_ref(&self) -> &url::Url { &self.url