use std::{
borrow::Cow,
future::Future,
net::SocketAddr,
num::NonZeroU32,
path::PathBuf,
pin::Pin,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
};
use derive_more::Debug;
use http::{
HeaderMap, HeaderValue, Method, Request, Response, StatusCode,
header::{AUTHORIZATION, InvalidHeaderValue},
response::Builder as ResponseBuilder,
};
use http_body_util::Full;
use hyper::body::Incoming;
use iroh_base::EndpointId;
#[cfg(feature = "test-utils")]
use iroh_base::RelayUrl;
use n0_error::{e, stack_error};
use n0_future::{StreamExt, task::AbortOnDropHandle};
use rustls::server::WantsServerCert;
use serde::Serialize;
use tokio::{
net::TcpListener,
task::{JoinError, JoinSet},
};
use tokio_rustls_acme::{
acme::{LETS_ENCRYPT_PRODUCTION_DIRECTORY, LETS_ENCRYPT_STAGING_DIRECTORY},
caches::DirCache,
};
use tracing::{Instrument, debug, error, info, info_span, instrument};
use self::http_server::{BytesBody, HyperError, HyperResult};
use crate::{
defaults::DEFAULT_KEY_CACHE_CAPACITY,
http::{AUTH_TOKEN_URL_QUERY_PARAM, ProtocolVersion, RELAY_PROBE_PATH},
quic::server::{QuicServer, QuicSpawnError, ServerHandle as QuicServerHandle},
tls::CaTlsConfig,
};
pub mod client;
pub mod clients;
pub mod http_server;
mod metrics;
pub(crate) mod resolver;
pub mod streams;
#[cfg(feature = "test-utils")]
pub mod testing;
pub use self::{
http_server::{Handlers, RelayService},
metrics::{Metrics, RelayMetrics},
resolver::{DEFAULT_CERT_RELOAD_INTERVAL, reloading_resolver},
};
const NO_CONTENT_CHALLENGE_HEADER: &str = "X-Iroh-Challenge";
const NO_CONTENT_RESPONSE_HEADER: &str = "X-Iroh-Response";
const NOTFOUND: &[u8] = b"Not Found";
const ROBOTS_TXT: &[u8] = b"User-agent: *\nDisallow: /\n";
const INDEX: &[u8] = br#"<html><body>
<h1>Iroh Relay</h1>
<p>
This is an <a href="https://iroh.computer/">Iroh</a> Relay server.
</p>
"#;
const TLS_HEADERS: [(&str, &str); 2] = [
(
"Strict-Transport-Security",
"max-age=63072000; includeSubDomains",
),
(
"Content-Security-Policy",
"default-src 'none'; frame-ancestors 'none'; form-action 'none'; base-uri 'self'; block-all-mixed-content; plugin-types 'none'",
),
];
fn body_empty() -> BytesBody {
Box::new(http_body_util::Full::new(hyper::body::Bytes::new()))
}
#[derive(Debug, Default)]
#[non_exhaustive]
pub struct ServerConfig {
pub relay: Option<RelayConfig>,
pub quic: Option<QuicConfig>,
#[cfg(feature = "metrics")]
pub metrics_addr: Option<SocketAddr>,
}
#[derive(Debug)]
#[non_exhaustive]
pub struct RelayConfig {
pub http_bind_addr: SocketAddr,
pub tls: Option<TlsConfig>,
pub limits: Limits,
pub key_cache_capacity: Option<usize>,
pub access: Arc<dyn DynAccessControl>,
}
impl RelayConfig {
pub fn new(http_bind_addr: impl Into<SocketAddr>) -> Self {
Self {
http_bind_addr: http_bind_addr.into(),
tls: None,
limits: Limits::default(),
key_cache_capacity: None,
access: Arc::new(AllowAll),
}
}
}
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, derive_more::Display)]
#[display("{_0}")]
pub struct ConnectionId(u64);
impl ConnectionId {
fn next() -> Self {
static NEXT: AtomicU64 = AtomicU64::new(0);
Self(NEXT.fetch_add(1, Ordering::Relaxed))
}
}
#[derive(Debug, Clone)]
pub struct ClientRequest {
connection_id: ConnectionId,
endpoint_id: EndpointId,
protocol_version: ProtocolVersion,
request: http::request::Parts,
}
impl ClientRequest {
pub fn new(
endpoint_id: EndpointId,
protocol_version: ProtocolVersion,
request: http::request::Parts,
) -> Self {
Self {
connection_id: ConnectionId::next(),
endpoint_id,
protocol_version,
request,
}
}
pub fn connection_id(&self) -> ConnectionId {
self.connection_id
}
pub fn protocol_version(&self) -> ProtocolVersion {
self.protocol_version
}
pub fn endpoint_id(&self) -> EndpointId {
self.endpoint_id
}
pub fn uri(&self) -> &http::Uri {
&self.request.uri
}
pub fn query_pairs(&self) -> impl Iterator<Item = (Cow<'_, str>, Cow<'_, str>)> {
url::form_urlencoded::parse(self.request.uri.query().unwrap_or("").as_bytes())
}
pub fn headers(&self) -> &http::HeaderMap {
&self.request.headers
}
pub fn auth_token(&self) -> Option<String> {
for value in self.request.headers.get_all(AUTHORIZATION) {
let value = value.to_str().ok()?;
if let Some((scheme, token)) = value.split_once(' ')
&& scheme.eq_ignore_ascii_case("Bearer")
{
return Some(token.to_string());
}
}
self.query_pairs()
.find(|(name, _)| name == AUTH_TOKEN_URL_QUERY_PARAM)
.map(|(_, value)| value.into_owned())
}
}
pub trait AccessControl: std::fmt::Debug + Send + Sync + 'static {
fn on_connect(&self, request: &ClientRequest) -> impl Future<Output = Access> + Send;
fn on_disconnect(&self, endpoint_id: EndpointId, connection_id: ConnectionId) {
let _ = (endpoint_id, connection_id);
}
}
pub trait DynAccessControl: std::fmt::Debug + Send + Sync + 'static {
fn on_connect<'a>(
&'a self,
request: &'a ClientRequest,
) -> Pin<Box<dyn Future<Output = Access> + Send + 'a>>;
fn on_disconnect(&self, endpoint_id: EndpointId, connection_id: ConnectionId);
}
impl<T: AccessControl> DynAccessControl for T {
fn on_connect<'a>(
&'a self,
request: &'a ClientRequest,
) -> Pin<Box<dyn Future<Output = Access> + Send + 'a>> {
Box::pin(<Self as AccessControl>::on_connect(self, request))
}
fn on_disconnect(&self, endpoint_id: EndpointId, connection_id: ConnectionId) {
<Self as AccessControl>::on_disconnect(self, endpoint_id, connection_id)
}
}
#[derive(Debug, Clone, Copy)]
pub struct AllowAll;
impl AccessControl for AllowAll {
async fn on_connect(&self, _request: &ClientRequest) -> Access {
Access::Allow
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Access {
Allow,
Deny {
reason: Option<String>,
},
}
#[derive(Debug)]
pub struct OnDisconnectGuard {
access: Option<Arc<dyn DynAccessControl>>,
endpoint_id: EndpointId,
connection_id: ConnectionId,
}
impl OnDisconnectGuard {
pub fn for_access_control(access: Arc<dyn DynAccessControl>, request: &ClientRequest) -> Self {
Self {
access: Some(access),
endpoint_id: request.endpoint_id(),
connection_id: request.connection_id(),
}
}
pub fn empty(endpoint_id: EndpointId) -> Self {
Self {
access: None,
endpoint_id,
connection_id: ConnectionId::next(),
}
}
pub fn endpoint_id(&self) -> EndpointId {
self.endpoint_id
}
pub fn connection_id(&self) -> ConnectionId {
self.connection_id
}
}
impl Drop for OnDisconnectGuard {
fn drop(&mut self) {
if let Some(access) = self.access.as_ref() {
access.on_disconnect(self.endpoint_id, self.connection_id);
}
}
}
#[derive(Debug)]
#[non_exhaustive]
pub struct QuicConfig {
pub bind_addr: SocketAddr,
pub server_config: Option<rustls::ServerConfig>,
}
impl QuicConfig {
pub fn new(bind_addr: impl Into<SocketAddr>) -> Self {
Self {
bind_addr: bind_addr.into(),
server_config: None,
}
}
}
#[derive(Debug)]
#[non_exhaustive]
pub struct TlsConfig {
pub https_bind_addr: SocketAddr,
pub cert: CertConfig,
}
impl TlsConfig {
pub fn new(https_bind_addr: impl Into<SocketAddr>, cert: CertConfig) -> Self {
Self {
https_bind_addr: https_bind_addr.into(),
cert,
}
}
}
#[derive(Debug, Default)]
#[non_exhaustive]
pub struct Limits {
pub client_rx: Option<ClientRateLimit>,
pub accept_conn_limit: Option<f64>,
pub accept_conn_burst: Option<usize>,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct ClientRateLimit {
pub bytes_per_second: NonZeroU32,
pub max_burst_bytes: Option<NonZeroU32>,
}
impl ClientRateLimit {
pub fn new(bytes_per_second: NonZeroU32) -> Self {
Self {
bytes_per_second,
max_burst_bytes: None,
}
}
}
#[derive(Debug)]
#[non_exhaustive]
pub enum CertConfig {
LetsEncrypt {
acme_config: AcmeConfig,
server_config_builder: rustls::ConfigBuilder<rustls::ServerConfig, WantsServerCert>,
},
Manual {
server_config: rustls::ServerConfig,
},
}
#[derive(Debug, Clone)]
pub struct AcmeConfig {
pub(crate) directory_url: String,
pub(crate) domains: Vec<String>,
pub(crate) contact: Vec<String>,
pub(crate) cache_path: Option<PathBuf>,
pub(crate) tls_config: CaTlsConfig,
}
impl AcmeConfig {
pub fn new(directory_url: String) -> Self {
Self {
directory_url,
domains: Vec::new(),
contact: Vec::new(),
cache_path: None,
tls_config: CaTlsConfig::default(),
}
}
pub fn letsencrypt(production: bool) -> Self {
let url = if production {
LETS_ENCRYPT_PRODUCTION_DIRECTORY
} else {
LETS_ENCRYPT_STAGING_DIRECTORY
};
Self::new(url.to_string())
}
pub fn domains(mut self, domains: Vec<String>) -> Self {
self.domains = domains;
self
}
pub fn contact(mut self, contact: Vec<String>) -> Self {
self.contact = contact;
self
}
pub fn cache_path(mut self, path: PathBuf) -> Self {
self.cache_path = Some(path);
self
}
pub fn tls_config(mut self, tls_config: CaTlsConfig) -> Self {
self.tls_config = tls_config;
self
}
}
#[derive(Debug)]
pub struct Server {
http_addr: Option<SocketAddr>,
https_addr: Option<SocketAddr>,
quic_addr: Option<SocketAddr>,
relay_handle: Option<http_server::ServerHandle>,
relay_service: Option<http_server::RelayService>,
quic_handle: Option<QuicServerHandle>,
supervisor: AbortOnDropHandle<Result<(), SupervisorError>>,
metrics: RelayMetrics,
metrics_server: Option<iroh_metrics::service::MetricsServer>,
}
#[allow(missing_docs)]
#[stack_error(derive, add_meta, std_sources)]
#[non_exhaustive]
pub enum SpawnError {
#[error("Unable to get local address")]
LocalAddr { source: std::io::Error },
#[error("Failed to bind QAD listener")]
QuicSpawn { source: QuicSpawnError },
#[error("Failed to parse TLS header")]
TlsHeaderParse { source: InvalidHeaderValue },
#[error("Failed to bind TcpListener")]
BindTlsListener { source: std::io::Error },
#[error("Failed to build ACME client TLS config")]
AcmeClientTlsConfig {
#[error(std_err)]
source: std::io::Error,
},
#[error("No local address")]
NoLocalAddr { source: std::io::Error },
#[error("Failed to bind server socket to {addr}")]
BindTcpListener {
source: std::io::Error,
addr: SocketAddr,
},
#[error("Error starting metrics server")]
Metrics {
#[error(std_err)]
source: std::io::Error,
},
}
#[allow(missing_docs)]
#[stack_error(derive, add_meta)]
#[non_exhaustive]
pub enum SupervisorError {
#[error("Acme event stream finished")]
AcmeEventStreamFinished {},
#[error(transparent)]
JoinError {
#[error(from, std_err)]
source: JoinError,
},
#[error("No relay services are enabled")]
NoRelayServicesEnabled {},
#[error("Task cancelled")]
TaskCancelled {},
}
impl Server {
pub async fn spawn(config: ServerConfig) -> Result<Self, SpawnError> {
let mut tasks = JoinSet::new();
let metrics = RelayMetrics::default();
#[cfg(not(feature = "metrics"))]
let metrics_server = None;
#[cfg(feature = "metrics")]
let metrics_server = if let Some(addr) = config.metrics_addr {
debug!("Starting metrics server");
let mut registry = iroh_metrics::Registry::default();
registry.register_all(&metrics);
let server = iroh_metrics::service::MetricsServer::spawn(addr, Arc::new(registry))
.await
.map_err(|err| e!(SpawnError::Metrics, err))?;
Some(server)
} else {
None
};
let (relay_server, http_addr, tls_config) = match config.relay {
Some(relay_config) => {
debug!("Starting Relay server");
let mut headers = HeaderMap::new();
for (name, value) in TLS_HEADERS.iter() {
headers.insert(
*name,
value
.parse()
.map_err(|err| e!(SpawnError::TlsHeaderParse, err))?,
);
}
let relay_bind_addr = match relay_config.tls {
Some(ref tls) => tls.https_bind_addr,
None => relay_config.http_bind_addr,
};
let key_cache_capacity = relay_config
.key_cache_capacity
.unwrap_or(DEFAULT_KEY_CACHE_CAPACITY);
let mut builder = http_server::ServerBuilder::new(relay_bind_addr)
.metrics(metrics.server.clone())
.headers(headers)
.key_cache_capacity(key_cache_capacity)
.access(relay_config.access)
.request_handler(Method::GET, "/", Box::new(root_handler))
.request_handler(Method::GET, "/index.html", Box::new(root_handler))
.request_handler(Method::GET, RELAY_PROBE_PATH, Box::new(probe_handler))
.request_handler(Method::GET, "/robots.txt", Box::new(robots_handler))
.request_handler(Method::GET, "/healthz", Box::new(healthz_handler));
if let Some(cfg) = relay_config.limits.client_rx {
builder = builder.client_rx_ratelimit(cfg);
}
let (http_addr, tls_config) = match relay_config.tls {
Some(tls_config) => {
let server_tls_config = match tls_config.cert {
CertConfig::LetsEncrypt {
acme_config,
server_config_builder,
} => {
let cache = acme_config.cache_path.map(DirCache::new);
let crypto_provider =
server_config_builder.crypto_provider().clone();
let client_tls_config = acme_config
.tls_config
.client_config(crypto_provider)
.map_err(|err| e!(SpawnError::AcmeClientTlsConfig, err))?;
let config =
tokio_rustls_acme::AcmeConfig::new_with_client_tls_config(
acme_config.domains,
Arc::new(client_tls_config),
)
.contact(acme_config.contact)
.directory(acme_config.directory_url)
.cache_option(cache);
let mut state = config.state();
let resolver = state.resolver().clone();
let server_config =
server_config_builder.with_cert_resolver(resolver);
let acceptor =
http_server::TlsAcceptor::LetsEncrypt(state.acceptor());
tasks.spawn(
async move {
while let Some(event) = state.next().await {
match event {
Ok(ok) => debug!("acme event: {ok:?}"),
Err(err) => error!("error: {err:?}"),
}
}
Err(e!(SupervisorError::AcmeEventStreamFinished))
}
.instrument(info_span!("acme")),
);
http_server::TlsConfig {
config: Arc::new(server_config),
acceptor,
}
}
CertConfig::Manual { server_config } => {
let server_config = Arc::new(server_config);
let acceptor =
tokio_rustls::TlsAcceptor::from(server_config.clone());
let acceptor = http_server::TlsAcceptor::Manual(acceptor);
http_server::TlsConfig {
config: server_config,
acceptor,
}
}
};
builder = builder.tls_config(Some(server_tls_config.clone()));
let http_listener = TcpListener::bind(&relay_config.http_bind_addr)
.await
.map_err(|err| e!(SpawnError::BindTlsListener, err))?;
let http_addr = http_listener
.local_addr()
.map_err(|err| e!(SpawnError::NoLocalAddr, err))?;
tasks.spawn(
async move {
run_captive_portal_service(http_listener).await;
Ok(())
}
.instrument(info_span!("http-service", addr = %http_addr)),
);
(Some(http_addr), Some(server_tls_config))
}
None => {
builder = builder.request_handler(
Method::GET,
"/generate_204",
Box::new(serve_no_content_handler),
);
(None, None)
}
};
let relay_server = builder.spawn().await?;
(Some(relay_server), http_addr, tls_config)
}
None => (None, None, None),
};
let relay_addr = relay_server.as_ref().map(|srv| srv.addr());
let relay_handle = relay_server.as_ref().map(|srv| srv.handle());
let relay_service = relay_server.as_ref().map(|srv| srv.service().clone());
let quic_server = match config.quic {
Some(quic_config) => {
debug!("Starting QUIC server {}", quic_config.bind_addr);
let server_config = quic_config
.server_config
.or(tls_config.map(|config| (*config.config).clone()))
.ok_or_else(|| {
e!(SpawnError::QuicSpawn, e!(QuicSpawnError::TlsNotConfigured))
})?;
Some(
QuicServer::spawn(quic_config.bind_addr, server_config, metrics.server.clone())
.map_err(|err| e!(SpawnError::QuicSpawn, err))?,
)
}
None => None,
};
let quic_addr = quic_server.as_ref().map(|srv| srv.bind_addr());
let quic_handle = quic_server.as_ref().map(|srv| srv.handle());
let task = tokio::spawn(relay_supervisor(tasks, relay_server, quic_server));
Ok(Self {
http_addr: http_addr.or(relay_addr),
https_addr: http_addr.and(relay_addr),
quic_addr,
relay_handle,
relay_service,
quic_handle,
supervisor: AbortOnDropHandle::new(task),
metrics,
metrics_server,
})
}
pub async fn shutdown(self) -> Result<(), SupervisorError> {
if let Some(handle) = self.relay_handle {
handle.shutdown();
}
if let Some(handle) = self.quic_handle {
handle.shutdown();
}
if let Some(server) = self.metrics_server {
server.shutdown().await;
}
self.supervisor.await?
}
pub async fn join(&mut self) -> Result<Result<(), SupervisorError>, JoinError> {
(&mut self.supervisor).await
}
pub fn https_addr(&self) -> Option<SocketAddr> {
self.https_addr
}
pub fn http_addr(&self) -> Option<SocketAddr> {
self.http_addr
}
pub fn quic_addr(&self) -> Option<SocketAddr> {
self.quic_addr
}
#[cfg(feature = "test-utils")]
pub fn https_url(&self) -> Option<RelayUrl> {
self.https_addr.map(|addr| {
url::Url::parse(&format!("https://{addr}"))
.expect("valid url")
.into()
})
}
#[cfg(feature = "test-utils")]
pub fn http_url(&self) -> Option<RelayUrl> {
self.http_addr.map(|addr| {
url::Url::parse(&format!("http://{addr}"))
.expect("valid url")
.into()
})
}
pub fn metrics(&self) -> &RelayMetrics {
&self.metrics
}
pub fn relay_service(&self) -> Option<&RelayService> {
self.relay_service.as_ref()
}
}
#[instrument(skip_all)]
async fn relay_supervisor(
mut tasks: JoinSet<Result<(), SupervisorError>>,
mut relay_http_server: Option<http_server::Server>,
mut quic_server: Option<QuicServer>,
) -> Result<(), SupervisorError> {
let quic_enabled = quic_server.is_some();
let mut quic_fut = match quic_server {
Some(ref mut server) => n0_future::Either::Left(server.task_handle()),
None => n0_future::Either::Right(n0_future::future::pending()),
};
let relay_enabled = relay_http_server.is_some();
let mut relay_fut = match relay_http_server {
Some(ref mut server) => n0_future::Either::Left(server.task_handle()),
None => n0_future::Either::Right(n0_future::future::pending()),
};
let res = tokio::select! {
biased;
Some(ret) = tasks.join_next() => ret,
ret = &mut quic_fut, if quic_enabled => ret.map(Ok),
ret = &mut relay_fut, if relay_enabled => ret.map(Ok),
else => Ok(Err(e!(SupervisorError::NoRelayServicesEnabled))),
};
let ret = match res {
Ok(Ok(())) => {
debug!("Task exited");
Ok(())
}
Ok(Err(err)) => {
error!(%err, "Task failed");
Err(err)
}
Err(err) => {
if let Ok(panic) = err.try_into_panic() {
error!("Task panicked");
std::panic::resume_unwind(panic);
}
debug!("Task cancelled");
Err(e!(SupervisorError::TaskCancelled))
}
};
if let Some(server) = relay_http_server {
server.shutdown();
}
if let Some(server) = quic_server {
server.shutdown().await;
}
tasks.shutdown().await;
ret
}
fn root_handler(
_r: Request<Incoming>,
response: ResponseBuilder,
) -> HyperResult<Response<BytesBody>> {
let body: BytesBody = Box::new(Full::from(INDEX));
response
.status(StatusCode::OK)
.header("Content-Type", "text/html; charset=utf-8")
.body(body)
.map_err(|err| Box::new(err) as HyperError)
}
fn probe_handler(
_r: Request<Incoming>,
response: ResponseBuilder,
) -> HyperResult<Response<BytesBody>> {
response
.status(StatusCode::OK)
.header("Access-Control-Allow-Origin", "*")
.body(body_empty())
.map_err(|err| Box::new(err) as HyperError)
}
fn robots_handler(
_r: Request<Incoming>,
response: ResponseBuilder,
) -> HyperResult<Response<BytesBody>> {
let body: BytesBody = Box::new(Full::from(ROBOTS_TXT));
response
.status(StatusCode::OK)
.body(body)
.map_err(|err| Box::new(err) as HyperError)
}
fn serve_no_content_handler<B: hyper::body::Body>(
r: Request<B>,
mut response: ResponseBuilder,
) -> HyperResult<Response<BytesBody>> {
let check = |c: &HeaderValue| {
!c.is_empty() && c.len() < 64 && c.as_bytes().iter().all(|c| is_challenge_char(*c as char))
};
if let Some(challenge) = r.headers().get(NO_CONTENT_CHALLENGE_HEADER)
&& check(challenge)
{
response = response.header(
NO_CONTENT_RESPONSE_HEADER,
format!("response {}", challenge.to_str()?),
);
}
response
.status(StatusCode::NO_CONTENT)
.body(body_empty())
.map_err(|err| Box::new(err) as HyperError)
}
fn is_challenge_char(c: char) -> bool {
c.is_ascii_lowercase()
|| c.is_ascii_uppercase()
|| c.is_ascii_digit()
|| c == '.'
|| c == '-'
|| c == '_'
}
#[derive(Serialize)]
struct Health {
status: &'static str,
version: &'static str,
git_hash: &'static str,
}
fn healthz_handler(
_r: Request<Incoming>,
response: ResponseBuilder,
) -> HyperResult<Response<BytesBody>> {
let health = Health {
status: "ok",
version: env!("CARGO_PKG_VERSION"),
git_hash: option_env!("VERGEN_GIT_SHA").unwrap_or("unknown"),
};
let body = serde_json::to_string(&health).unwrap_or_else(|_| r#"{"status":"error"}"#.into());
let body: BytesBody = Box::new(Full::from(body));
response
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(body)
.map_err(|err| Box::new(err) as HyperError)
}
async fn run_captive_portal_service(http_listener: TcpListener) {
info!("serving");
let mut tasks = JoinSet::new();
loop {
tokio::select! {
biased;
Some(res) = tasks.join_next() => {
if let Err(err) = res
&& err.is_panic()
{
panic!("task panicked: {err:#?}");
}
}
res = http_listener.accept() => {
match res {
Ok((stream, peer_addr)) => {
debug!(%peer_addr, "Connection opened",);
let handler = CaptivePortalService;
tasks.spawn(async move {
let stream = crate::server::streams::MaybeTlsStream::Plain(stream);
let stream = hyper_util::rt::TokioIo::new(stream);
if let Err(err) = hyper::server::conn::http1::Builder::new()
.serve_connection(stream, handler)
.with_upgrades()
.await
{
error!("Failed to serve connection: {err:?}");
}
});
}
Err(err) => {
error!(
"[CaptivePortalService] failed to accept connection: {:#?}",
err
);
}
}
}
}
}
}
#[derive(Clone)]
struct CaptivePortalService;
impl hyper::service::Service<Request<Incoming>> for CaptivePortalService {
type Response = Response<BytesBody>;
type Error = HyperError;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn call(&self, req: Request<Incoming>) -> Self::Future {
match (req.method(), req.uri().path()) {
(&Method::GET, "/generate_204") => {
Box::pin(async move { serve_no_content_handler(req, Response::builder()) })
}
_ => {
let body: BytesBody = Box::new(Full::from(NOTFOUND));
let r = Response::builder()
.status(StatusCode::NOT_FOUND)
.body(body)
.map_err(|err| Box::new(err) as HyperError);
Box::pin(async move { r })
}
}
}
}
#[cfg(test)]
mod tests {
use std::{net::Ipv4Addr, sync::Arc, time::Duration};
use http::StatusCode;
use iroh_base::{EndpointId, RelayUrl, SecretKey};
use iroh_dns::dns::DnsResolver;
use n0_error::{Result, StackResultExt, StdResultExt};
use n0_future::{SinkExt, StreamExt};
use n0_tracing_test::traced_test;
use rand::{RngExt, SeedableRng};
use tracing::{info, instrument};
use url::Url;
use super::{
Access, AccessControl, ClientRequest, NO_CONTENT_CHALLENGE_HEADER,
NO_CONTENT_RESPONSE_HEADER, RelayConfig, Server, ServerConfig, SpawnError,
};
use crate::{
client::{ClientBuilder, ConnectError},
protos::{
handshake,
relay::{ClientToRelayMsg, Datagrams, RelayToClientMsg},
},
test_utils::static_resolver,
tls::{self, CaTlsConfig, default_provider},
};
#[derive(derive_more::Debug)]
struct TestAccess(#[debug("access fn")] Box<dyn Fn(&ClientRequest) -> Access + Send + Sync>);
impl AccessControl for TestAccess {
async fn on_connect(&self, request: &ClientRequest) -> Access {
(self.0)(request)
}
}
async fn spawn_local_relay() -> std::result::Result<Server, SpawnError> {
let mut relay = RelayConfig::new((Ipv4Addr::LOCALHOST, 0));
relay.key_cache_capacity = Some(1024);
Server::spawn(ServerConfig {
relay: Some(relay),
quic: None,
metrics_addr: None,
})
.await
}
#[instrument]
async fn try_send_recv(
client_a: &mut crate::client::Client,
client_b: &mut crate::client::Client,
b_key: EndpointId,
msg: Datagrams,
) -> Result<RelayToClientMsg> {
for _ in 0..10 {
client_a
.send(ClientToRelayMsg::Datagrams {
dst_endpoint_id: b_key,
datagrams: msg.clone(),
})
.await?;
let Ok(res) = tokio::time::timeout(Duration::from_millis(500), client_b.next()).await
else {
continue;
};
let res = res.expect("stream finished")?;
return Ok(res);
}
panic!("failed to send and recv message");
}
fn dns_resolver() -> DnsResolver {
DnsResolver::new()
}
fn ring_config() -> rustls::ClientConfig {
rustls::ClientConfig::builder_with_provider(Arc::new(
rustls::crypto::ring::default_provider(),
))
.with_safe_default_protocol_versions()
.unwrap()
.with_root_certificates(rustls::RootCertStore::empty())
.with_no_client_auth()
}
#[tokio::test]
#[traced_test]
async fn test_no_services() {
let mut server = Server::spawn(ServerConfig::default()).await.unwrap();
let res = tokio::time::timeout(Duration::from_secs(5), server.join())
.await
.expect("timeout, server not finished")
.expect("server task JoinError");
assert!(res.is_err());
}
#[tokio::test]
#[traced_test]
async fn test_conflicting_bind() {
let mut relay = RelayConfig::new((Ipv4Addr::LOCALHOST, 1234));
relay.key_cache_capacity = Some(1024);
let res = Server::spawn(ServerConfig {
relay: Some(relay),
quic: None,
metrics_addr: Some((Ipv4Addr::LOCALHOST, 1234).into()),
})
.await;
assert!(res.is_err()); }
#[tokio::test]
#[traced_test]
async fn test_root_handler() {
let server = spawn_local_relay().await.unwrap();
let url = format!("http://{}", server.http_addr().unwrap());
let client = reqwest::Client::builder()
.use_preconfigured_tls(ring_config())
.build()
.unwrap();
let response = client.get(&url).send().await.unwrap();
assert_eq!(response.status(), 200);
let body = response.text().await.unwrap();
assert!(body.contains("iroh.computer"));
}
#[tokio::test]
#[traced_test]
async fn test_captive_portal_service() {
let server = spawn_local_relay().await.unwrap();
let url = format!("http://{}/generate_204", server.http_addr().unwrap());
let challenge = "123az__.";
let client = reqwest::Client::builder()
.use_preconfigured_tls(ring_config())
.build()
.unwrap();
let response = client
.get(&url)
.header(NO_CONTENT_CHALLENGE_HEADER, challenge)
.send()
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NO_CONTENT);
let header = response.headers().get(NO_CONTENT_RESPONSE_HEADER).unwrap();
assert_eq!(header.to_str().unwrap(), format!("response {challenge}"));
let body = response.text().await.unwrap();
assert!(body.is_empty());
}
#[tokio::test]
#[traced_test]
async fn test_relay_clients() -> Result<()> {
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let server = spawn_local_relay().await?;
let relay_url = format!("http://{}", server.http_addr().unwrap());
let relay_url: RelayUrl = relay_url.parse()?;
let client_config = CaTlsConfig::default()
.client_config(default_provider())
.unwrap();
let a_secret_key = SecretKey::from_bytes(&rng.random());
let a_key = a_secret_key.public();
let resolver = dns_resolver();
info!("client a build & connect");
let mut client_a = ClientBuilder::new(relay_url.clone(), a_secret_key, resolver.clone())
.tls_client_config(client_config.clone())
.connect()
.await?;
let b_secret_key = SecretKey::from_bytes(&rng.random());
let b_key = b_secret_key.public();
info!("client b build & connect");
let mut client_b = ClientBuilder::new(relay_url.clone(), b_secret_key, resolver.clone())
.tls_client_config(client_config)
.connect()
.await?;
info!("sending a -> b");
let msg = Datagrams::from("hello, b");
let res = try_send_recv(&mut client_a, &mut client_b, b_key, msg.clone()).await?;
let RelayToClientMsg::Datagrams {
remote_endpoint_id,
datagrams,
} = res
else {
panic!("client_b received unexpected message {res:?}");
};
assert_eq!(a_key, remote_endpoint_id);
assert_eq!(msg, datagrams);
info!("sending b -> a");
let msg = Datagrams::from("howdy, a");
let res = try_send_recv(&mut client_b, &mut client_a, a_key, msg.clone()).await?;
let RelayToClientMsg::Datagrams {
remote_endpoint_id,
datagrams,
} = res
else {
panic!("client_a received unexpected message {res:?}");
};
assert_eq!(b_key, remote_endpoint_id);
assert_eq!(msg, datagrams);
Ok(())
}
#[tokio::test]
#[traced_test]
async fn test_relay_access_control() -> Result<()> {
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let current_span = tracing::info_span!("this is a test");
let _guard = current_span.enter();
let client_config = CaTlsConfig::default()
.client_config(default_provider())
.unwrap();
let a_secret_key = SecretKey::from_bytes(&rng.random());
let a_key = a_secret_key.public();
let mut relay = RelayConfig::new((Ipv4Addr::LOCALHOST, 0));
relay.key_cache_capacity = Some(1024);
relay.access = Arc::new(TestAccess(Box::new(move |request| {
let endpoint_id = request.endpoint_id();
info!("checking {}", endpoint_id);
if endpoint_id == a_key {
Access::Deny { reason: None }
} else {
Access::Allow
}
})));
let server = Server::spawn(ServerConfig {
relay: Some(relay),
quic: None,
metrics_addr: None,
})
.await?;
let relay_url = format!("http://{}", server.http_addr().unwrap());
let relay_url: RelayUrl = relay_url.parse()?;
let resolver = dns_resolver();
let result = ClientBuilder::new(relay_url.clone(), a_secret_key, resolver)
.tls_client_config(client_config.clone())
.connect()
.await;
assert!(
matches!(result, Err(ConnectError::Handshake { source: handshake::Error::ServerDeniedAuth { reason, .. }, .. }) if reason == "not authorized")
);
let b_secret_key = SecretKey::from_bytes(&rng.random());
let b_key = b_secret_key.public();
let resolver = dns_resolver();
let mut client_b = ClientBuilder::new(relay_url.clone(), b_secret_key, resolver)
.tls_client_config(client_config.clone())
.connect()
.await?;
let c_secret_key = SecretKey::from_bytes(&rng.random());
let c_key = c_secret_key.public();
let resolver = dns_resolver();
let mut client_c = ClientBuilder::new(relay_url.clone(), c_secret_key, resolver)
.tls_client_config(client_config)
.connect()
.await?;
let msg = Datagrams::from("hello, c");
let res = try_send_recv(&mut client_b, &mut client_c, c_key, msg.clone()).await?;
if let RelayToClientMsg::Datagrams {
remote_endpoint_id,
datagrams,
} = res
{
assert_eq!(b_key, remote_endpoint_id);
assert_eq!(msg, datagrams);
} else {
panic!("client_c received unexpected message {res:?}");
}
Ok(())
}
#[tokio::test]
#[traced_test]
async fn test_relay_client_auth_token_forwarded() -> Result<()> {
const TOKEN: &str = "secret-token";
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let client_config = CaTlsConfig::default()
.client_config(default_provider())
.unwrap();
let mut relay = RelayConfig::new((Ipv4Addr::LOCALHOST, 0));
relay.key_cache_capacity = Some(1024);
relay.access = Arc::new(TestAccess(Box::new(move |request| {
if request.auth_token().as_deref() == Some(TOKEN) {
Access::Allow
} else {
Access::Deny { reason: None }
}
})));
let server = Server::spawn(ServerConfig {
relay: Some(relay),
quic: None,
metrics_addr: None,
})
.await?;
let relay_url = format!("http://{}", server.http_addr().unwrap());
let relay_url: RelayUrl = relay_url.parse()?;
let secret_key = SecretKey::from_bytes(&rng.random());
let result = ClientBuilder::new(relay_url.clone(), secret_key, dns_resolver())
.tls_client_config(client_config.clone())
.connect()
.await;
assert!(matches!(
result,
Err(ConnectError::Handshake { source: handshake::Error::ServerDeniedAuth { reason, .. }, .. })
if reason == "not authorized"
));
let secret_key = SecretKey::from_bytes(&rng.random());
let result = ClientBuilder::new(relay_url.clone(), secret_key, dns_resolver())
.tls_client_config(client_config.clone())
.auth_token("wrong-token")
.connect()
.await;
assert!(matches!(
result,
Err(ConnectError::Handshake { source: handshake::Error::ServerDeniedAuth { reason, .. }, .. })
if reason == "not authorized"
));
let secret_key = SecretKey::from_bytes(&rng.random());
let _client = ClientBuilder::new(relay_url, secret_key, dns_resolver())
.tls_client_config(client_config)
.auth_token(TOKEN)
.connect()
.await?;
Ok(())
}
#[tokio::test]
#[traced_test]
async fn test_relay_clients_full() -> Result<()> {
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let server = spawn_local_relay().await.unwrap();
let relay_url = format!("http://{}", server.http_addr().unwrap());
let relay_url: RelayUrl = relay_url.parse().unwrap();
let client_config = CaTlsConfig::default()
.client_config(default_provider())
.unwrap();
let a_secret_key = SecretKey::from_bytes(&rng.random());
let resolver = dns_resolver();
let mut client_a = ClientBuilder::new(relay_url.clone(), a_secret_key, resolver.clone())
.tls_client_config(client_config.clone())
.connect()
.await?;
let b_secret_key = SecretKey::from_bytes(&rng.random());
let b_key = b_secret_key.public();
let _client_b = ClientBuilder::new(relay_url.clone(), b_secret_key, resolver.clone())
.tls_client_config(client_config)
.connect()
.await?;
let msg = Datagrams::from("hello, b");
for _i in 0..1000 {
client_a
.send(ClientToRelayMsg::Datagrams {
dst_endpoint_id: b_key,
datagrams: msg.clone(),
})
.await?;
}
Ok(())
}
#[tokio::test]
#[traced_test]
async fn test_relay_client_falls_back_to_ipv4() -> Result {
let config = ServerConfig {
relay: Some(RelayConfig::new((Ipv4Addr::LOCALHOST, 0))),
..Default::default()
};
let server = Server::spawn(config).await?;
let addr = server.http_addr().expect("http relay address");
let resolver = static_resolver(
vec![Ipv4Addr::LOCALHOST],
vec!["2001:db8::dead".parse().expect("valid IPv6")],
);
let url: Url = format!("http://relay.test:{}", addr.port())
.parse()
.expect("valid relay url");
let client = ClientBuilder::new(url, SecretKey::generate(), resolver)
.tls_client_config(tls::make_dangerous_client_config())
.address_family_selector(|| true);
tokio::time::timeout(Duration::from_secs(10), client.connect())
.await
.with_std_context(|_| "relay connect timed out")?
.context("relay connect")?;
server.shutdown().await.context("relay server shutdown")?;
Ok(())
}
}