mod priority_connect;
use std::{borrow::Cow, fmt, net, sync::Arc, time::Duration};
use endhost_api_client::client::CrpcEndhostApiClient;
use rand::seq::IndexedRandom;
use reqwest_connect_rpc::{
client::CrpcClientError,
token_source::{TokenSource, static_token::StaticTokenSource},
};
use scion_sdk_utils::backoff::ExponentialBackoff;
use url::Url;
use x25519_dalek::StaticSecret;
pub use crate::underlays::udp::{OutboundIpResolver, TargetAddrOutboundIpResolver};
use crate::{
ea_source::{
EndhostApiSource, EndhostApiSourceError, StaticEndhostApiDiscovery, StaticEndhostApis,
},
path::fetcher::{EndhostApiSegmentFetcher, traits::SegmentFetcher},
stack::ScionStack,
underlays::{
SnapSocketConfig, UnderlayStack,
discovery::{PeriodicUnderlayDiscovery, UnderlayDiscovery},
},
};
const DEFAULT_UDP_NEXT_HOP_RESOLVER_FETCH_INTERVAL: Duration = Duration::from_secs(600);
const DEFAULT_ENDHOST_API_DISCOVERY_MAX_GROUPS: usize = 5;
const DEFAULT_ENDHOST_API_DISCOVERY_APIS_PER_GROUP: usize = 2;
const DEFAULT_ENDHOST_API_DISCOVERY_PER_GROUP_DELAY: Duration = Duration::from_millis(500);
type OutboundIpResolverFactory = Box<dyn FnOnce(Url) -> Arc<dyn OutboundIpResolver> + Send>;
pub struct ScionStackBuilder {
crpc_client: Option<reqwest::Client>,
endhost_api_token_source: Option<Arc<dyn TokenSource>>,
auth_token_source: Option<Arc<dyn TokenSource>>,
endhost_api_source: Arc<dyn EndhostApiSource>,
preferred_underlay: PreferredUnderlay,
endhost_api_discovery: EndhostApiDiscoveryConfig,
snap: SnapUnderlayConfig,
udp: UdpUnderlayConfig,
}
impl ScionStackBuilder {
#[must_use]
pub fn new() -> Self {
Self {
crpc_client: None,
endhost_api_token_source: None,
auth_token_source: None,
endhost_api_source: Arc::new(StaticEndhostApiDiscovery::global()),
preferred_underlay: PreferredUnderlay::Udp,
endhost_api_discovery: EndhostApiDiscoveryConfig::default(),
snap: SnapUnderlayConfig::default(),
udp: UdpUnderlayConfig::default(),
}
}
#[must_use]
pub fn with_preferred_underlay(mut self, preferred: PreferredUnderlay) -> Self {
self.preferred_underlay = preferred;
self
}
#[must_use]
pub fn with_crpc_client(mut self, crpc_client: reqwest::Client) -> Self {
self.crpc_client = Some(crpc_client);
self
}
#[must_use]
pub fn with_endhost_api(mut self, endhost_api_url: Url) -> Self {
let source = StaticEndhostApis::new().add_group(vec![endhost_api_url]);
self.endhost_api_source = Arc::new(source);
self
}
#[must_use]
pub fn with_endhost_api_discovery_source(mut self, source: impl EndhostApiSource) -> Self {
self.endhost_api_source = Arc::new(source);
self
}
#[must_use]
pub fn with_endhost_api_auth_token_source(mut self, source: impl TokenSource) -> Self {
self.endhost_api_token_source = Some(Arc::new(source));
self
}
#[must_use]
pub fn with_endhost_api_auth_token(mut self, token: String) -> Self {
self.endhost_api_token_source = Some(Arc::new(StaticTokenSource::from(token)));
self
}
#[must_use]
pub fn with_auth_token_source(mut self, source: impl TokenSource) -> Self {
self.auth_token_source = Some(Arc::new(source));
self
}
#[must_use]
pub fn with_auth_token(mut self, token: String) -> Self {
self.auth_token_source = Some(Arc::new(StaticTokenSource::from(token)));
self
}
#[must_use]
pub fn with_endhost_api_discovery_max_groups(mut self, max_groups: usize) -> Self {
self.endhost_api_discovery.max_groups = max_groups;
self
}
#[must_use]
pub fn with_anapaya_ead_apis_per_group(mut self, apis_per_group: usize) -> Self {
self.endhost_api_discovery.apis_per_group = apis_per_group;
self
}
#[must_use]
pub fn with_endhost_api_discovery_per_group_delay(mut self, per_group_delay: Duration) -> Self {
self.endhost_api_discovery.per_group_delay = per_group_delay;
self
}
#[must_use]
pub fn with_snap_underlay_config(mut self, config: SnapUnderlayConfig) -> Self {
self.snap = config;
self
}
#[must_use]
pub fn with_udp_underlay_config(mut self, config: UdpUnderlayConfig) -> Self {
self.udp = config;
self
}
pub async fn build(self) -> Result<ScionStack, BuildScionStackError> {
let ScionStackBuilder {
crpc_client,
endhost_api_token_source,
auth_token_source,
endhost_api_source,
preferred_underlay,
endhost_api_discovery,
snap,
udp,
} = self;
let api_groups = endhost_api_source.endhost_apis().await?;
let api_groups: Vec<Vec<Url>> = {
let mut rng = rand::rng();
api_groups
.into_iter()
.map(|g| g.apis.into_iter().map(|a| a.address).collect::<Vec<_>>())
.filter(|group| !group.is_empty())
.take(endhost_api_discovery.max_groups)
.map(|group: Vec<Url>| {
group
.sample(&mut rng, endhost_api_discovery.apis_per_group)
.cloned()
.collect()
})
.collect()
};
if api_groups.is_empty() {
return Err(BuildScionStackError::EndhostApiSourceError(
EndhostApiSourceError::new("endhost API discovery returned no APIs", false),
));
}
let token_source: Option<Arc<dyn TokenSource>> =
endhost_api_token_source.or(auth_token_source.clone());
let crpc_c = crpc_client.clone();
let discover_underlays = move |url: Url| {
let token_source = token_source.clone();
let crpc_c = crpc_c.clone();
let url = url.clone();
async move {
let mut client = match crpc_c {
Some(client) => {
CrpcEndhostApiClient::new_with_client(&url, client)
.map_err(ApiAttemptError::client_setup)?
}
None => {
CrpcEndhostApiClient::new(&url).map_err(ApiAttemptError::client_setup)?
}
};
if let Some(token_source) = &token_source {
client.use_token_source(token_source.clone());
}
let client = Arc::new(client);
let discovery = PeriodicUnderlayDiscovery::new(
client.clone(),
udp.udp_next_hop_resolver_fetch_interval,
ExponentialBackoff::new(0.5, 10.0, 2.0, 0.5),
)
.await
.map_err(ApiAttemptError::underlay_discovery)?;
Ok((client, discovery))
}
};
let (api_url, (endhost_api_client, underlay_discovery)) =
priority_connect::try_priority_groups(
api_groups,
discover_underlays,
endhost_api_discovery.per_group_delay,
)
.await
.map_err(|errors| {
BuildScionStackError::AllEndhostApisFailed(AllEndhostApisFailed::new(errors))
})?;
tracing::info!(url=%api_url, "Selected endhost API");
let outbound_ip_resolver: Arc<dyn OutboundIpResolver> =
(udp.outbound_ip_resolver_factory)(api_url.clone());
let underlay_stack = UnderlayStack::new(
preferred_underlay,
Arc::new(underlay_discovery),
outbound_ip_resolver,
snap.static_identity.unwrap_or_else(StaticSecret::random),
SnapSocketConfig {
crpc_client: snap.crpc_client.or(crpc_client),
snap_token_source: snap.snap_token_source.or(auth_token_source),
},
);
Ok(ScionStack::new(
Some(api_url),
Arc::new(EndhostApiSegmentFetcher::new(endhost_api_client)),
Arc::new(underlay_stack),
))
}
fn build_static_udp_underlay(
underlay_discovery: Arc<dyn UnderlayDiscovery>,
outbound_ip_resolver: Arc<dyn OutboundIpResolver>,
default_segment_fetcher: Arc<dyn SegmentFetcher>,
) -> ScionStack {
let underlay_stack = UnderlayStack::new(
PreferredUnderlay::Udp,
underlay_discovery,
outbound_ip_resolver,
StaticSecret::random(),
SnapSocketConfig {
crpc_client: None,
snap_token_source: None,
},
);
ScionStack::new(None, default_segment_fetcher, Arc::new(underlay_stack))
}
}
impl Default for ScionStackBuilder {
fn default() -> Self {
Self::new()
}
}
impl ScionStack {
#[must_use]
pub fn static_udp_underlay(
underlay_discovery: Arc<dyn UnderlayDiscovery>,
outbound_ip_resolver: Arc<dyn OutboundIpResolver>,
default_segment_fetcher: Arc<dyn SegmentFetcher>,
) -> ScionStack {
ScionStackBuilder::build_static_udp_underlay(
underlay_discovery,
outbound_ip_resolver,
default_segment_fetcher,
)
}
}
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum BuildScionStackError {
#[error("no underlay available: {0}")]
UnderlayUnavailable(Cow<'static, str>),
#[error(transparent)]
AllEndhostApisFailed(#[from] AllEndhostApisFailed),
#[error(transparent)]
EndhostApiSourceError(#[from] EndhostApiSourceError),
#[error(transparent)]
Snap(#[from] BuildSnapScionStackError),
#[error("internal error")]
Internal(#[source] Box<dyn std::error::Error + Send + Sync>),
}
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum BuildSnapScionStackError {
#[error("no SNAP data plane available: {0}")]
DataPlaneUnavailable(Cow<'static, str>),
#[error("control plane client setup error")]
ControlPlaneClientSetup(#[source] Box<dyn std::error::Error + Send + Sync>),
#[error("data plane discovery request error")]
DataPlaneDiscovery(#[source] Box<dyn std::error::Error + Send + Sync>),
}
#[derive(Debug)]
pub struct AllEndhostApisFailed {
failures: Vec<(Url, ApiAttemptError)>,
}
impl AllEndhostApisFailed {
pub(crate) fn new(failures: Vec<(Url, ApiAttemptError)>) -> Self {
Self { failures }
}
#[must_use]
pub fn failures(&self) -> &[(Url, ApiAttemptError)] {
&self.failures
}
#[must_use]
pub fn is_transient(&self) -> bool {
!self.failures.is_empty() && self.failures.iter().all(|(_, err)| err.is_transient())
}
}
impl fmt::Display for AllEndhostApisFailed {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "all {} endhost API(s) failed", self.failures.len())?;
let mut sep = ": ";
for (url, err) in &self.failures {
write!(f, "{sep}{url} ({err})")?;
sep = "; ";
}
Ok(())
}
}
impl std::error::Error for AllEndhostApisFailed {}
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum ApiAttemptError {
#[error("client setup")]
ClientSetup {
transient: bool,
#[source]
source: Box<dyn std::error::Error + Send + Sync>,
},
#[error("underlay discovery")]
UnderlayDiscovery {
transient: bool,
#[source]
source: Box<dyn std::error::Error + Send + Sync>,
},
}
impl ApiAttemptError {
pub(crate) fn client_setup(error: anyhow::Error) -> Self {
Self::ClientSetup {
transient: false,
source: error.into_boxed_dyn_error(),
}
}
pub(crate) fn underlay_discovery(error: CrpcClientError) -> Self {
Self::UnderlayDiscovery {
transient: is_transient_crpc_error(&error),
source: Box::new(error),
}
}
#[must_use]
pub fn is_transient(&self) -> bool {
match self {
Self::ClientSetup { transient, .. } | Self::UnderlayDiscovery { transient, .. } => {
*transient
}
}
}
}
fn is_transient_crpc_error(error: &CrpcClientError) -> bool {
matches!(error, CrpcClientError::ConnectionError { .. })
}
pub struct EndhostApiDiscoveryConfig {
max_groups: usize,
apis_per_group: usize,
per_group_delay: Duration,
}
impl Default for EndhostApiDiscoveryConfig {
fn default() -> Self {
Self {
max_groups: DEFAULT_ENDHOST_API_DISCOVERY_MAX_GROUPS,
apis_per_group: DEFAULT_ENDHOST_API_DISCOVERY_APIS_PER_GROUP,
per_group_delay: DEFAULT_ENDHOST_API_DISCOVERY_PER_GROUP_DELAY,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum PreferredUnderlay {
Snap,
Udp,
}
#[derive(Default)]
pub struct SnapUnderlayConfig {
crpc_client: Option<reqwest::Client>,
snap_token_source: Option<Arc<dyn TokenSource>>,
snap_dp_index: usize,
static_identity: Option<StaticSecret>,
}
impl SnapUnderlayConfig {
#[must_use]
pub fn with_auth_token(mut self, token: String) -> Self {
self.snap_token_source = Some(Arc::new(StaticTokenSource::from(token)));
self
}
#[must_use]
pub fn with_auth_token_source(mut self, source: impl TokenSource) -> Self {
self.snap_token_source = Some(Arc::new(source));
self
}
#[must_use]
pub fn with_crpc_client(mut self, client: reqwest::Client) -> Self {
self.crpc_client = Some(client);
self
}
#[must_use]
pub fn with_snap_dp_index(mut self, dp_index: usize) -> Self {
self.snap_dp_index = dp_index;
self
}
#[must_use]
pub fn with_static_identity(mut self, identity: StaticSecret) -> Self {
self.static_identity = Some(identity);
self
}
}
pub struct UdpUnderlayConfig {
udp_next_hop_resolver_fetch_interval: Duration,
outbound_ip_resolver_factory: OutboundIpResolverFactory,
}
impl Default for UdpUnderlayConfig {
fn default() -> Self {
Self {
udp_next_hop_resolver_fetch_interval: DEFAULT_UDP_NEXT_HOP_RESOLVER_FETCH_INTERVAL,
outbound_ip_resolver_factory: Box::new(move |url| {
Arc::new(TargetAddrOutboundIpResolver::new(url, vec![]))
}),
}
}
}
impl UdpUnderlayConfig {
#[must_use]
pub fn with_outbound_ips(mut self, outbound_ips: Vec<net::IpAddr>) -> Self {
self.outbound_ip_resolver_factory =
Box::new(move |_url| Arc::new(outbound_ips) as Arc<dyn OutboundIpResolver>);
self
}
#[must_use]
pub fn with_outbound_ip_resolver(
mut self,
resolver: impl OutboundIpResolver + 'static,
) -> Self {
let resolver = Arc::new(resolver) as Arc<dyn OutboundIpResolver>;
self.outbound_ip_resolver_factory = Box::new(move |_url| resolver.clone());
self
}
#[must_use]
pub fn with_outbound_ip_resolver_factory<F, R>(mut self, factory: F) -> Self
where
F: FnOnce(Url) -> R + Send + 'static,
R: OutboundIpResolver + 'static,
{
self.outbound_ip_resolver_factory =
Box::new(move |url| Arc::new(factory(url)) as Arc<dyn OutboundIpResolver>);
self
}
#[must_use]
pub fn with_udp_next_hop_resolver_fetch_interval(mut self, fetch_interval: Duration) -> Self {
self.udp_next_hop_resolver_fetch_interval = fetch_interval;
self
}
}
#[cfg(test)]
mod tests {
use std::borrow::Cow;
use reqwest_connect_rpc::client::CrpcClientError;
use url::Url;
use super::*;
fn connection_error() -> CrpcClientError {
CrpcClientError::ConnectionError {
context: Cow::Borrowed("test"),
source: Box::new(std::io::Error::other("boom")),
}
}
fn non_connection_error() -> CrpcClientError {
CrpcClientError::DecodeError {
context: Cow::Borrowed("test"),
source: Some(Box::new(std::io::Error::other("boom"))),
body: None,
}
}
#[test]
fn api_attempt_error_transient_classification() {
assert!(ApiAttemptError::underlay_discovery(connection_error()).is_transient());
assert!(!ApiAttemptError::underlay_discovery(non_connection_error()).is_transient());
assert!(!ApiAttemptError::client_setup(anyhow::anyhow!("invalid url")).is_transient());
}
#[test]
fn all_endhost_apis_failed_transient_classification() {
let url: Url = "http://example.com".parse().expect("valid url");
assert!(!AllEndhostApisFailed::new(vec![]).is_transient());
assert!(
AllEndhostApisFailed::new(vec![(
url.clone(),
ApiAttemptError::underlay_discovery(connection_error()),
)])
.is_transient()
);
assert!(
!AllEndhostApisFailed::new(vec![
(
url.clone(),
ApiAttemptError::underlay_discovery(connection_error()),
),
(
url,
ApiAttemptError::underlay_discovery(non_connection_error())
),
])
.is_transient()
);
}
}