GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HTTP & DNS Improvements (#6423)

Browse Source 
* 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)
This commit is contained in:
Jack Wampler
2026-02-11 07:04:53 -07:00
committed by GitHub
parent 5fc2936d3f
commit e7fcaa980f
7 changed files with 908 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/http-api-client/src/dns.rs
+214 -82
View File
@@ -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<R: Resolve + 'static>(mut self, resolver: Arc<R>) -> 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<OnceCell<TokioResolver>>,
fallback: Option<Arc<OnceCell<TokioResolver>>>,
use_system: Arc<AtomicBool>,
system_resolver: Arc<OnceCell<TokioResolver>>,
static_base: Option<Arc<OnceCell<StaticResolver>>>,
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<Addrs, Box<dyn std::error::Error + Send + Sync>> {
Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
}
async fn resolve(
name: Name,
resolver: Arc<OnceCell<TokioResolver>>,
maybe_fallback: Option<Arc<OnceCell<TokioResolver>>>,
resolver: TokioResolver,
maybe_static: Option<Arc<OnceCell<StaticResolver>>>,
independent: bool,
overall_dns_timeout: Duration,
) -> Result<Addrs, ResolveError> {
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<impl Iterator<Item = IpAddr> + 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<TokioResolver, ResolveError> {
// 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<HashMap<String, Vec<IpAddr>>> {
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<String, Vec<IpAddr>>) {
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<String, Vec<IpAddr>>) {
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<HashMap<String, Vec<IpAddr>>> {
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<String, Vec<IpAddr>>) {
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<TokioResolver, ResolveError> {
}
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(())
}
}
}
common/http-api-client/src/dns/constants.rs
+6
View File
@@ -72,6 +72,12 @@ pub const NYM_RPC_IPS: &[IpAddr] = &[
)),
];
#[allow(unused)]
pub fn empty_static_addrs() -> HashMap<String, Vec<IpAddr>> {
HashMap::new()
}
#[allow(unused)]
pub fn default_static_addrs() -> HashMap<String, Vec<IpAddr>> {
let mut m = HashMap::new();
m.insert(NYM_API_DOMAIN.to_string(), NYM_API_IPS.to_vec());
common/http-api-client/src/dns/static_resolver.rs
+279 -52
View File
@@ -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<Mutex<HashMap<String, Entry>>>,
fallback_addr_map: Arc<Mutex<HashMap<String, Vec<IpAddr>>>>,
preresolve_addr_map: Arc<Mutex<HashMap<String, Entry>>>,
pre_resolve_timeout: Option<Duration>,
}
#[derive(Debug, Clone, Default)]
enum PreResolveStatus {
#[default]
Valid,
ValidUntil(Instant),
}
#[derive(Debug, Clone, Default)]
struct Entry {
valid_for_pre_resolve_until: Option<Instant>,
status: PreResolveStatus,
addrs: Vec<IpAddr>,
}
impl Entry {
fn new(addrs: Vec<IpAddr>) -> Self {
Self {
valid_for_pre_resolve_until: None,
status: PreResolveStatus::Valid,
addrs,
}
}
fn new_timeout(addrs: Vec<IpAddr>, 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<String, Vec<IpAddr>>) -> 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<String, Vec<IpAddr>> {
/// 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<String, Vec<IpAddr>>) -> 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<String, Vec<IpAddr>>) -> 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<String, Vec<IpAddr>> {
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<String, Vec<IpAddr>> {
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<String, Vec<IpAddr>>) {
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<String, Vec<IpAddr>>) {
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<Vec<IpAddr>> {
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<Vec<IpAddr>> {
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<String, Entry>>,
fallback_table: MutexGuard<'_, HashMap<String, Vec<IpAddr>>>,
mut preresolve_table: MutexGuard<'_, HashMap<String, Entry>>,
name: &str,
timeout: Option<Duration>,
) -> Option<Entry> {
let resolved = table.get_mut(name)?;
pre_resolve_cache_timeout: Option<Duration>,
) -> Option<Vec<IpAddr>> {
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<PreResolveStatus> {
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)
}
}
}
common/http-api-client/src/fronted.rs
+217 -16
View File
@@ -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<Arc<RwLock<FrontPolicy>>> =
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<RwLock<FrontPolicy>>,
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<FrontPolicy>) -> 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");
common/http-api-client/src/lib.rs
+179 -90
View File
@@ -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<reqwest::Client> = 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<reqwest::Url>,
@@ -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<Url>);
/// 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<Url>,
timeout: Option<Duration>,
custom_user_agent: bool,
reqwest_client_builder: reqwest::ClientBuilder,
custom_user_agent: Option<HeaderValue>,
reqwest_client_builder: Option<reqwest::ClientBuilder>,
#[allow(dead_code)] // not dead code, just unused in wasm
use_secure_dns: bool,
#[cfg(feature = "tunneling")]
front: Option<fronted::Front>,
front: fronted::Front,
retry_limit: usize,
serialization: SerializationFormat,
error: Option<HttpClientError>,
}
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<HeaderValue>,
V::Error: Into<http::Error>,
{
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<Client, HttpClientError> {
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<Url>,
current_idx: Arc<AtomicUsize>,
reqwest_client: reqwest::Client,
using_secure_dns: bool,
reqwest_client: Option<reqwest::Client>,
custom_user_agent: Option<HeaderValue>,
#[cfg(feature = "tunneling")]
front: Option<fronted::Front>,
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<Response, HttpClientError> = {
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<Url>) {
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<T>(
&self,
res: Response,
allow_empty: bool,
) -> Result<T, HttpClientError>
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
}
}
common/http-api-client/src/tests.rs
+3 -2
View File
@@ -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<dyn std::error::Error>> {
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();
common/http-api-client/src/url.rs
+10
View File
@@ -123,6 +123,16 @@ impl From<reqwest::Url> 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<url::Url> for Url {
fn as_ref(&self) -> &url::Url {
&self.url