#[cfg(not(wasm_browser))]
use std::net::{SocketAddrV4, SocketAddrV6};
use std::{
collections::BTreeSet,
net::{IpAddr, SocketAddr},
sync::Arc,
};
use http::StatusCode;
use iroh_base::RelayUrl;
use iroh_relay::{RelayMap, RelayNode, defaults::DEFAULT_RELAY_QUIC_PORT, http::RELAY_PROBE_PATH};
#[cfg(not(wasm_browser))]
use iroh_relay::{
dns::{DnsError, DnsResolver, StaggeredError},
quic::QuicClient,
};
#[cfg(wasm_browser)]
use n0_future::future::Pending;
use n0_future::{
StreamExt as _,
task::{self, AbortOnDropHandle, JoinSet},
time::{self, Duration, Instant},
};
use rand::seq::IteratorRandom;
use snafu::{IntoError, OptionExt, ResultExt, Snafu};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{Instrument, debug, debug_span, error, info_span, trace, warn};
use url::Host;
#[cfg(wasm_browser)]
use super::portmapper; use super::{
Report,
probes::{Probe, ProbePlan},
};
#[cfg(not(wasm_browser))]
use super::{defaults::timeouts::DNS_TIMEOUT, ip_mapped_addrs::IpMappedAddresses};
#[cfg(not(wasm_browser))]
use crate::discovery::dns::DNS_STAGGERING_MS;
use crate::net_report::defaults::timeouts::{
CAPTIVE_PORTAL_DELAY, CAPTIVE_PORTAL_TIMEOUT, OVERALL_REPORT_TIMEOUT, PROBES_TIMEOUT,
};
#[derive(Debug)]
pub(super) struct Client {
_drop_guard: AbortOnDropHandle<()>,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct IfStateDetails {
pub have_v4: bool,
pub have_v6: bool,
}
impl IfStateDetails {
#[cfg(test)]
pub(super) fn fake() -> Self {
IfStateDetails {
have_v4: true,
have_v6: true,
}
}
}
impl From<netwatch::netmon::State> for IfStateDetails {
fn from(value: netwatch::netmon::State) -> Self {
IfStateDetails {
have_v4: value.have_v4,
have_v6: value.have_v6,
}
}
}
#[cfg(not(wasm_browser))]
#[derive(Debug, Clone)]
pub(crate) struct SocketState {
pub(crate) quic_client: Option<QuicClient>,
pub(crate) dns_resolver: DnsResolver,
pub(crate) ip_mapped_addrs: Option<IpMappedAddresses>,
}
impl Client {
pub(super) fn new(
last_report: Option<Report>,
relay_map: RelayMap,
protocols: BTreeSet<Probe>,
if_state: IfStateDetails,
#[cfg(not(wasm_browser))] socket_state: SocketState,
#[cfg(any(test, feature = "test-utils"))] insecure_skip_relay_cert_verify: bool,
) -> (Self, mpsc::Receiver<ProbeFinished>) {
let (msg_tx, msg_rx) = mpsc::channel(32);
let actor = Actor {
msg_tx,
last_report,
relay_map,
protocols,
#[cfg(not(wasm_browser))]
socket_state,
#[cfg(any(test, feature = "test-utils"))]
insecure_skip_relay_cert_verify,
if_state,
};
let task = task::spawn(actor.run().instrument(info_span!("reportgen.actor")));
(
Self {
_drop_guard: AbortOnDropHandle::new(task),
},
msg_rx,
)
}
}
#[derive(Debug)]
struct Actor {
msg_tx: mpsc::Sender<ProbeFinished>,
last_report: Option<Report>,
relay_map: RelayMap,
protocols: BTreeSet<Probe>,
#[cfg(not(wasm_browser))]
socket_state: SocketState,
#[cfg(any(test, feature = "test-utils"))]
insecure_skip_relay_cert_verify: bool,
if_state: IfStateDetails,
}
#[allow(missing_docs)]
#[derive(Debug, Snafu)]
#[non_exhaustive]
#[snafu(module)]
pub(super) enum ProbesError {
#[snafu(display("Probe failed"))]
ProbeFailure { source: ProbeError },
#[snafu(display("All probes failed"))]
AllProbesFailed,
#[snafu(display("Probe cancelled"))]
Cancelled,
#[snafu(display("Probe timed out"))]
Timeout,
}
#[derive(Debug)]
pub(super) enum ProbeFinished {
Regular(Result<ProbeReport, ProbesError>),
#[cfg(not(wasm_browser))]
CaptivePortal(Option<bool>),
}
impl Actor {
async fn run(self) {
match time::timeout(OVERALL_REPORT_TIMEOUT, self.run_inner()).await {
Ok(()) => debug!("reportgen actor finished"),
Err(time::Elapsed { .. }) => {
warn!("reportgen timed out");
}
}
}
async fn run_inner(self) {
debug!("reportstate actor starting");
let mut probes = JoinSet::default();
let _probes_token = self.spawn_probes_task(self.if_state.clone(), &mut probes);
let mut num_probes = probes.len();
let captive_token = self.prepare_captive_portal_task(&mut probes);
let mut have_udp = false;
while let Some(probe_result) = probes.join_next().await {
trace!(?probe_result, num_probes, "processing finished probe");
match probe_result {
Ok(report) => {
#[cfg_attr(wasm_browser, allow(irrefutable_let_patterns))]
if let ProbeFinished::Regular(report) = &report {
have_udp |= report.as_ref().map(|r| r.is_udp()).unwrap_or_default();
num_probes -= 1;
if num_probes == 0 {
debug!("all regular probes done");
debug_assert!(probes.len() <= 1, "{} probes", probes.len());
if have_udp {
captive_token.cancel();
}
}
}
self.msg_tx.send(report).await.ok();
}
Err(e) => {
if e.is_panic() {
error!("Task panicked {:?}", e);
break;
}
warn!("probes task join error: {:?}", e);
}
}
}
}
fn prepare_captive_portal_task(&self, tasks: &mut JoinSet<ProbeFinished>) -> CancellationToken {
let token = CancellationToken::new();
#[cfg(not(wasm_browser))]
if self.last_report.is_none() {
let preferred_relay = self
.last_report
.as_ref()
.and_then(|l| l.preferred_relay.clone());
let dns_resolver = self.socket_state.dns_resolver.clone();
let dm = self.relay_map.clone();
let token = token.clone();
tasks.spawn(
async move {
let res = token
.run_until_cancelled_owned(async move {
time::sleep(CAPTIVE_PORTAL_DELAY).await;
trace!("check started after {CAPTIVE_PORTAL_DELAY:?}");
time::timeout(
CAPTIVE_PORTAL_TIMEOUT,
check_captive_portal(&dns_resolver, &dm, preferred_relay),
)
.await
})
.await;
let res = match res {
Some(Ok(Ok(found))) => Some(found),
Some(Ok(Err(err))) => {
match err {
CaptivePortalError::CreateReqwestClient { source }
| CaptivePortalError::HttpRequest { source }
if source.is_connect() =>
{
debug!("check_captive_portal failed: {source:#}");
}
err => warn!("check_captive_portal error: {err:#}"),
}
None
}
Some(Err(time::Elapsed { .. })) => {
warn!("probe timed out");
None
}
None => {
trace!("probe cancelled");
None
}
};
ProbeFinished::CaptivePortal(res)
}
.instrument(debug_span!("captive-portal")),
);
}
token
}
fn spawn_probes_task(
&self,
if_state: IfStateDetails,
probes: &mut JoinSet<ProbeFinished>,
) -> CancellationToken {
debug!(?if_state, "local interface details");
let plan = match self.last_report {
Some(ref report) => {
ProbePlan::with_last_report(&self.relay_map, report, &self.protocols)
}
None => ProbePlan::initial(&self.relay_map, &self.protocols),
};
trace!(%plan, "probe plan");
let token = CancellationToken::new();
for probe_set in plan.iter() {
let set_token = token.child_token();
let proto = probe_set.proto();
for (delay, relay_node) in probe_set.params() {
let probe_token = set_token.child_token();
let fut = probe_token.run_until_cancelled_owned(time::timeout(
PROBES_TIMEOUT,
proto.run(
*delay,
relay_node.clone(),
#[cfg(not(wasm_browser))]
self.socket_state.clone(),
#[cfg(any(test, feature = "test-utils"))]
self.insecure_skip_relay_cert_verify,
),
));
probes.spawn(
async move {
let res = fut.await;
let res = match res {
Some(Ok(Ok(report))) => Ok(report),
Some(Ok(Err(err))) => {
warn!("probe failed: {:#}", err);
Err(probes_error::ProbeFailureSnafu {}.into_error(err))
}
Some(Err(time::Elapsed { .. })) => {
Err(probes_error::TimeoutSnafu.build())
}
None => Err(probes_error::CancelledSnafu.build()),
};
ProbeFinished::Regular(res)
}
.instrument(debug_span!(
"run-probe",
?proto,
?delay,
?relay_node
)),
);
}
}
token
}
}
#[derive(Debug, Clone)]
pub(super) enum ProbeReport {
#[cfg(not(wasm_browser))]
QadIpv4(QadProbeReport),
#[cfg(not(wasm_browser))]
QadIpv6(QadProbeReport),
Https(HttpsProbeReport),
}
impl ProbeReport {
#[cfg(not(wasm_browser))]
pub(super) fn is_udp(&self) -> bool {
matches!(self, Self::QadIpv4(_) | Self::QadIpv6(_))
}
#[cfg(wasm_browser)]
pub(super) fn is_udp(&self) -> bool {
false
}
}
#[cfg(not(wasm_browser))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct QadProbeReport {
pub(super) node: RelayUrl,
pub(super) latency: Duration,
pub(super) addr: SocketAddr,
}
#[derive(Debug, Clone)]
pub(super) struct HttpsProbeReport {
pub(super) node: RelayUrl,
pub(super) latency: Duration,
}
#[allow(missing_docs)]
#[derive(Debug, Snafu)]
#[snafu(module)]
#[non_exhaustive]
pub(super) enum ProbeError {
#[snafu(display("Client is gone"))]
ClientGone,
#[snafu(display("Probe is no longer useful"))]
NotUseful,
#[snafu(display("Failed to run HTTPS probe"))]
Https { source: MeasureHttpsLatencyError },
}
#[allow(missing_docs)]
#[derive(Debug, Snafu)]
#[snafu(module)]
#[non_exhaustive]
pub(super) enum QuicError {
#[snafu(display("No QUIC endpoint available"))]
NoEndpoint,
#[snafu(display("URL must have 'host' to use QUIC address discovery probes"))]
InvalidUrl,
}
#[derive(derive_more::Debug, Clone)]
pub struct QuicConfig {
#[debug("quinn::Endpoint")]
pub ep: quinn::Endpoint,
pub client_config: rustls::ClientConfig,
pub ipv4: bool,
pub ipv6: bool,
}
impl Probe {
async fn run(
self,
delay: Duration,
relay_node: Arc<RelayNode>,
#[cfg(not(wasm_browser))] socket_state: SocketState,
#[cfg(any(test, feature = "test-utils"))] insecure_skip_relay_cert_verify: bool,
) -> Result<ProbeReport, ProbeError> {
if !delay.is_zero() {
trace!("delaying probe");
time::sleep(delay).await;
}
debug!("starting probe");
match self {
Probe::Https => {
match run_https_probe(
#[cfg(not(wasm_browser))]
&socket_state.dns_resolver,
relay_node.url.clone(),
#[cfg(any(test, feature = "test-utils"))]
insecure_skip_relay_cert_verify,
)
.await
{
Ok(report) => Ok(ProbeReport::Https(report)),
Err(err) => Err(probe_error::HttpsSnafu.into_error(err)),
}
}
#[cfg(not(wasm_browser))]
Probe::QadIpv4 | Probe::QadIpv6 => unreachable!("must not be used"),
}
}
}
#[cfg(not(wasm_browser))]
pub(super) fn maybe_to_mapped_addr(
ip_mapped_addrs: Option<&IpMappedAddresses>,
addr: SocketAddr,
) -> SocketAddr {
if let Some(ip_mapped_addrs) = ip_mapped_addrs {
return ip_mapped_addrs.get_or_register(addr).private_socket_addr();
}
addr
}
#[cfg(not(wasm_browser))]
#[derive(Debug, Snafu)]
#[snafu(module)]
#[non_exhaustive]
enum CaptivePortalError {
#[snafu(transparent)]
DnsLookup { source: StaggeredError<DnsError> },
#[snafu(display("Creating HTTP client failed"))]
CreateReqwestClient { source: reqwest::Error },
#[snafu(display("HTTP request failed"))]
HttpRequest { source: reqwest::Error },
}
#[cfg(not(wasm_browser))]
async fn check_captive_portal(
dns_resolver: &DnsResolver,
dm: &RelayMap,
preferred_relay: Option<RelayUrl>,
) -> Result<bool, CaptivePortalError> {
let preferred_relay = preferred_relay.and_then(|url| dm.get_node(&url).map(|_| url));
let url = match preferred_relay {
Some(url) => url,
None => {
let urls: Vec<_> = dm.nodes().map(|n| n.url.clone()).collect();
if urls.is_empty() {
debug!("No suitable relay node for captive portal check");
return Ok(false);
}
let i = (0..urls.len())
.choose(&mut rand::thread_rng())
.unwrap_or_default();
urls[i].clone()
}
};
let mut builder = reqwest::ClientBuilder::new().redirect(reqwest::redirect::Policy::none());
if let Some(Host::Domain(domain)) = url.host() {
let addrs: Vec<_> = dns_resolver
.lookup_ipv4_ipv6_staggered(domain, DNS_TIMEOUT, DNS_STAGGERING_MS)
.await?
.map(|ipaddr| SocketAddr::new(ipaddr, 0))
.collect();
builder = builder.resolve_to_addrs(domain, &addrs);
}
let client = builder
.build()
.context(captive_portal_error::CreateReqwestClientSnafu)?;
let host_name = url.host_str().unwrap_or_default();
let challenge = format!("ts_{host_name}");
let portal_url = format!("http://{host_name}/generate_204");
let res = client
.request(reqwest::Method::GET, portal_url)
.header("X-Tailscale-Challenge", &challenge)
.send()
.await
.context(captive_portal_error::HttpRequestSnafu)?;
let expected_response = format!("response {challenge}");
let is_valid_response = res
.headers()
.get("X-Tailscale-Response")
.map(|s| s.to_str().unwrap_or_default())
== Some(&expected_response);
debug!(
"check_captive_portal url={} status_code={} valid_response={}",
res.url(),
res.status(),
is_valid_response,
);
let has_captive = res.status() != 204 || !is_valid_response;
Ok(has_captive)
}
#[cfg(not(wasm_browser))]
fn get_quic_port(relay_node: &RelayNode) -> Option<u16> {
if let Some(ref quic) = relay_node.quic {
if quic.port == 0 {
Some(DEFAULT_RELAY_QUIC_PORT)
} else {
Some(quic.port)
}
} else {
None
}
}
#[cfg(not(wasm_browser))]
#[derive(Debug, Snafu)]
#[snafu(module)]
#[non_exhaustive]
pub enum GetRelayAddrError {
#[snafu(display("No valid hostname in the relay URL"))]
InvalidHostname,
#[snafu(display("No suitable relay address found"))]
NoAddrFound,
#[snafu(display("DNS lookup failed"))]
DnsLookup { source: StaggeredError<DnsError> },
#[snafu(display("Relay node is not suitable for non-STUN probes"))]
UnsupportedRelayNode,
#[snafu(display("HTTPS probes are not implemented"))]
UnsupportedHttps,
#[snafu(display("No port available for this protocol"))]
MissingPort,
}
#[cfg(not(wasm_browser))]
pub(super) async fn get_relay_addr_ipv4(
dns_resolver: &DnsResolver,
relay_node: &RelayNode,
) -> Result<SocketAddrV4, GetRelayAddrError> {
let port = get_quic_port(relay_node).context(get_relay_addr_error::MissingPortSnafu)?;
relay_lookup_ipv4_staggered(dns_resolver, relay_node, port).await
}
#[cfg(not(wasm_browser))]
pub(super) async fn get_relay_addr_ipv6(
dns_resolver: &DnsResolver,
relay_node: &RelayNode,
) -> Result<SocketAddrV6, GetRelayAddrError> {
let port = get_quic_port(relay_node).context(get_relay_addr_error::MissingPortSnafu)?;
relay_lookup_ipv6_staggered(dns_resolver, relay_node, port).await
}
#[cfg(not(wasm_browser))]
async fn relay_lookup_ipv4_staggered(
dns_resolver: &DnsResolver,
relay: &RelayNode,
port: u16,
) -> Result<SocketAddrV4, GetRelayAddrError> {
match relay.url.host() {
Some(url::Host::Domain(hostname)) => {
trace!(%hostname, "Performing DNS A lookup for relay addr");
match dns_resolver
.lookup_ipv4_staggered(hostname, DNS_TIMEOUT, DNS_STAGGERING_MS)
.await
{
Ok(mut addrs) => addrs
.next()
.map(|ip| ip.to_canonical())
.map(|addr| match addr {
IpAddr::V4(ip) => SocketAddrV4::new(ip, port),
IpAddr::V6(_) => unreachable!("bad DNS lookup: {:?}", addr),
})
.ok_or(get_relay_addr_error::NoAddrFoundSnafu.build()),
Err(err) => Err(get_relay_addr_error::DnsLookupSnafu.into_error(err)),
}
}
Some(url::Host::Ipv4(addr)) => Ok(SocketAddrV4::new(addr, port)),
Some(url::Host::Ipv6(_addr)) => Err(get_relay_addr_error::NoAddrFoundSnafu.build()),
None => Err(get_relay_addr_error::InvalidHostnameSnafu.build()),
}
}
#[cfg(not(wasm_browser))]
async fn relay_lookup_ipv6_staggered(
dns_resolver: &DnsResolver,
relay: &RelayNode,
port: u16,
) -> Result<SocketAddrV6, GetRelayAddrError> {
match relay.url.host() {
Some(url::Host::Domain(hostname)) => {
trace!(%hostname, "Performing DNS AAAA lookup for relay addr");
match dns_resolver
.lookup_ipv6_staggered(hostname, DNS_TIMEOUT, DNS_STAGGERING_MS)
.await
{
Ok(mut addrs) => addrs
.next()
.map(|addr| match addr {
IpAddr::V4(_) => unreachable!("bad DNS lookup: {:?}", addr),
IpAddr::V6(ip) => SocketAddrV6::new(ip, port, 0, 0),
})
.ok_or(get_relay_addr_error::NoAddrFoundSnafu.build()),
Err(err) => Err(get_relay_addr_error::DnsLookupSnafu.into_error(err)),
}
}
Some(url::Host::Ipv4(_addr)) => Err(get_relay_addr_error::NoAddrFoundSnafu.build()),
Some(url::Host::Ipv6(addr)) => Ok(SocketAddrV6::new(addr, port, 0, 0)),
None => Err(get_relay_addr_error::InvalidHostnameSnafu.build()),
}
}
#[derive(Debug, Snafu)]
#[snafu(module)]
#[non_exhaustive]
pub enum MeasureHttpsLatencyError {
#[snafu(transparent)]
InvalidUrl { source: url::ParseError },
#[cfg(not(wasm_browser))]
#[snafu(transparent)]
DnsLookup { source: StaggeredError<DnsError> },
#[snafu(display("Creating HTTP client failed"))]
CreateReqwestClient { source: reqwest::Error },
#[snafu(display("HTTP request failed"))]
HttpRequest { source: reqwest::Error },
#[snafu(display("Error response from server {status}: {:?}", status.canonical_reason()))]
InvalidResponse { status: StatusCode },
}
#[allow(clippy::unused_async)]
async fn run_https_probe(
#[cfg(not(wasm_browser))] dns_resolver: &DnsResolver,
relay_node: RelayUrl,
#[cfg(any(test, feature = "test-utils"))] insecure_skip_relay_cert_verify: bool,
) -> Result<HttpsProbeReport, MeasureHttpsLatencyError> {
trace!("HTTPS probe start");
let url = relay_node.join(RELAY_PROBE_PATH)?;
let mut builder = reqwest::ClientBuilder::new();
#[cfg(not(wasm_browser))]
{
builder = builder.redirect(reqwest::redirect::Policy::none());
}
#[cfg(not(wasm_browser))]
if let Some(Host::Domain(domain)) = url.host() {
let addrs: Vec<_> = dns_resolver
.lookup_ipv4_ipv6_staggered(domain, DNS_TIMEOUT, DNS_STAGGERING_MS)
.await?
.map(|ipaddr| SocketAddr::new(ipaddr, 0))
.collect();
trace!(?addrs, "resolved addrs");
builder = builder.resolve_to_addrs(domain, &addrs);
}
#[cfg(all(not(wasm_browser), any(test, feature = "test-utils")))]
let builder = builder.danger_accept_invalid_certs(insecure_skip_relay_cert_verify);
let client = builder
.build()
.context(measure_https_latency_error::CreateReqwestClientSnafu)?;
let start = Instant::now();
let response = client
.request(reqwest::Method::GET, url)
.send()
.await
.context(measure_https_latency_error::HttpRequestSnafu)?;
let latency = start.elapsed();
if response.status().is_success() {
const MAX_BODY_SIZE: usize = 8 << 10; let mut body_size = 0;
let mut stream = response.bytes_stream();
while let Some(Ok(chunk)) = stream.next().await {
body_size += chunk.len();
if body_size >= MAX_BODY_SIZE {
break;
}
}
Ok(HttpsProbeReport {
node: relay_node,
latency,
})
} else {
Err(measure_https_latency_error::InvalidResponseSnafu {
status: response.status(),
}
.build())
}
}
#[cfg(test)]
mod tests {
use std::net::Ipv4Addr;
use iroh_relay::dns::DnsResolver;
use n0_snafu::{Result, ResultExt};
use tracing_test::traced_test;
use super::{super::test_utils, *};
#[tokio::test]
async fn test_measure_https_latency() -> Result {
let (_server, relay) = test_utils::relay().await;
let dns_resolver = DnsResolver::new();
tracing::info!(relay_url = ?relay.url , "RELAY_URL");
let report = run_https_probe(&dns_resolver, relay.url, true).await?;
assert!(report.latency > Duration::ZERO);
Ok(())
}
#[tokio::test]
#[traced_test]
async fn test_qad_probe_v4() -> Result {
let (server, relay) = test_utils::relay().await;
let relay = Arc::new(relay);
let client_config = iroh_relay::client::make_dangerous_client_config();
let ep = quinn::Endpoint::client(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0)).e()?;
let client_addr = ep.local_addr().e()?;
let quic_client = iroh_relay::quic::QuicClient::new(ep.clone(), client_config);
let dns_resolver = DnsResolver::default();
let (report, conn) = super::super::run_probe_v4(None, relay, quic_client, dns_resolver)
.await
.unwrap();
assert_eq!(report.addr, client_addr);
drop(conn);
ep.wait_idle().await;
server.shutdown().await?;
Ok(())
}
}