pub mod builder;
pub mod scmp_handler;
pub mod socket;
use std::{borrow::Cow, fmt, net, sync::Arc, time::Duration};
use async_trait::async_trait;
use futures::future::BoxFuture;
use sciparse::{
address::ip_socket_addr::ScionSocketIpAddr,
identifier::{isd::Isd, isd_asn::IsdAsn},
packet::view::ScionRawPacketView,
};
pub use socket::{PathUnawareUdpScionSocket, RawScionSocket, ScmpScionSocket, UdpScionSocket};
use url::Url;
pub use self::builder::ScionStackBuilder;
use crate::{
internal::Subscribers,
path::{
PathStrategy,
fetcher::{PathFetcherImpl, traits::SegmentFetcher},
manager::{
MultiPathManager, MultiPathManagerConfig,
traits::{PathWaitError, PathWaitTimeoutError},
},
policy::PathPolicy,
},
stack::{
scmp_handler::{ScmpErrorHandler, ScmpErrorReceiver},
socket::SendErrorReceiver,
},
};
pub struct ScionStack {
endhost_api: Option<Url>,
default_segment_fetcher: Arc<dyn SegmentFetcher>,
underlay: Arc<dyn DynUnderlayStack>,
scmp_error_receivers: Subscribers<dyn ScmpErrorReceiver>,
send_error_receivers: Subscribers<dyn SendErrorReceiver>,
}
#[allow(clippy::missing_fields_in_debug)]
impl fmt::Debug for ScionStack {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ScionStack")
.field("endhost_api", &self.endhost_api)
.finish()
}
}
impl ScionStack {
pub(crate) fn new(
endhost_api: Option<Url>,
default_segment_fetcher: Arc<dyn SegmentFetcher>,
underlay: Arc<dyn DynUnderlayStack>,
) -> Self {
Self {
endhost_api,
default_segment_fetcher,
underlay,
scmp_error_receivers: Subscribers::new(),
send_error_receivers: Subscribers::new(),
}
}
pub async fn bind(
&self,
bind_addr: Option<ScionSocketIpAddr>,
) -> Result<UdpScionSocket, ScionSocketBindError> {
self.bind_with_config(bind_addr, SocketConfig::default())
.await
}
pub async fn bind_with_config(
&self,
bind_addr: Option<ScionSocketIpAddr>,
mut socket_config: SocketConfig,
) -> Result<UdpScionSocket, ScionSocketBindError> {
let socket = PathUnawareUdpScionSocket::new(
self.underlay
.bind_socket(SocketKind::Udp, bind_addr)
.await?,
vec![Box::new(ScmpErrorHandler::new(
self.scmp_error_receivers.clone(),
))],
);
if !socket_config.disable_default_segment_fetcher {
socket_config
.segment_fetchers
.push(("Endhost API".into(), self.default_segment_fetcher.clone()));
}
let fetcher = PathFetcherImpl::new(
socket_config.segment_fetchers,
socket_config.segment_fetcher_timeout,
);
if socket_config.path_strategy.scoring.is_empty() {
socket_config.path_strategy.scoring.use_default_scorers();
}
let pather = Arc::new(
MultiPathManager::new(
MultiPathManagerConfig::default(),
fetcher,
socket_config.path_strategy,
)
.expect("should not fail with default configuration"),
);
self.scmp_error_receivers.register(pather.clone());
self.send_error_receivers.register(pather.clone());
Ok(UdpScionSocket::new(
socket,
pather,
socket_config.connect_timeout,
self.send_error_receivers.clone(),
))
}
pub async fn connect(
&self,
remote_addr: ScionSocketIpAddr,
bind_addr: Option<ScionSocketIpAddr>,
) -> Result<UdpScionSocket, ScionSocketConnectError> {
let socket = self.bind(bind_addr).await?;
socket.connect(remote_addr).await
}
pub async fn connect_with_config(
&self,
remote_addr: ScionSocketIpAddr,
bind_addr: Option<ScionSocketIpAddr>,
socket_config: SocketConfig,
) -> Result<UdpScionSocket, ScionSocketConnectError> {
let socket = self.bind_with_config(bind_addr, socket_config).await?;
socket.connect(remote_addr).await
}
pub async fn bind_scmp(
&self,
bind_addr: Option<ScionSocketIpAddr>,
) -> Result<ScmpScionSocket, ScionSocketBindError> {
let socket = self
.underlay
.bind_socket(SocketKind::Scmp, bind_addr)
.await?;
Ok(ScmpScionSocket::new(socket))
}
pub async fn bind_raw(
&self,
bind_addr: Option<ScionSocketIpAddr>,
) -> Result<RawScionSocket, ScionSocketBindError> {
let socket = self
.underlay
.bind_socket(SocketKind::Raw, bind_addr)
.await?;
Ok(RawScionSocket::new(socket))
}
pub async fn bind_path_unaware(
&self,
bind_addr: Option<ScionSocketIpAddr>,
) -> Result<PathUnawareUdpScionSocket, ScionSocketBindError> {
let socket = self
.underlay
.bind_socket(SocketKind::Udp, bind_addr)
.await?;
Ok(PathUnawareUdpScionSocket::new(socket, vec![]))
}
pub fn local_ases(&self) -> Vec<IsdAsn> {
self.underlay.local_ases()
}
pub fn endhost_api(&self) -> Option<Url> {
self.endhost_api.clone()
}
pub fn create_path_manager(&self) -> MultiPathManager<PathFetcherImpl> {
let fetcher = PathFetcherImpl::new(
vec![("Endhost API".into(), self.default_segment_fetcher.clone())],
DEFAULT_SEGMENT_FETCHER_TIMEOUT,
);
let mut strategy = PathStrategy::default();
strategy.scoring.use_default_scorers();
MultiPathManager::new(MultiPathManagerConfig::default(), fetcher, strategy)
.expect("should not fail with default configuration")
}
pub fn create_path_fetcher(&self) -> PathFetcherImpl {
PathFetcherImpl::new(
vec![("Endhost API".into(), self.default_segment_fetcher.clone())],
DEFAULT_SEGMENT_FETCHER_TIMEOUT,
)
}
}
pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
pub const DEFAULT_SEGMENT_FETCHER_TIMEOUT: Duration = Duration::from_secs(60);
pub struct SocketConfig {
pub(crate) segment_fetchers: Vec<(String, Arc<dyn SegmentFetcher>)>,
pub(crate) segment_fetcher_timeout: Duration,
pub(crate) disable_default_segment_fetcher: bool,
pub(crate) path_strategy: PathStrategy,
pub(crate) connect_timeout: Duration,
}
impl Default for SocketConfig {
fn default() -> Self {
Self::new()
}
}
impl SocketConfig {
#[must_use]
pub fn new() -> Self {
Self {
segment_fetchers: Vec::new(),
segment_fetcher_timeout: DEFAULT_SEGMENT_FETCHER_TIMEOUT,
disable_default_segment_fetcher: false,
path_strategy: PathStrategy::default(),
connect_timeout: DEFAULT_CONNECT_TIMEOUT,
}
}
#[must_use]
pub fn with_path_policy(mut self, policy: impl PathPolicy) -> Self {
self.path_strategy.add_policy(policy);
self
}
#[must_use]
pub fn with_connection_timeout(mut self, timeout: Duration) -> Self {
self.connect_timeout = timeout;
self
}
#[must_use]
pub fn with_segment_fetcher(mut self, name: String, fetcher: Arc<dyn SegmentFetcher>) -> Self {
self.segment_fetchers.push((name, fetcher));
self
}
#[must_use]
pub fn disable_default_segment_fetcher(mut self) -> Self {
self.disable_default_segment_fetcher = true;
self
}
#[must_use]
pub fn with_segment_fetcher_timeout(mut self, timeout: Duration) -> Self {
self.segment_fetcher_timeout = timeout;
self
}
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ScionSocketBindError {
#[error(transparent)]
InvalidBindAddress(InvalidBindAddressError),
#[error("port {0} is already in use")]
PortAlreadyInUse(u16),
#[error(transparent)]
SnapConnectionError(SnapConnectionError),
#[error("underlay unavailable for the requested ISD: {0}")]
NoUnderlayAvailable(Isd),
#[error("other error: {0}")]
Other(#[from] Box<dyn std::error::Error + Send + Sync>),
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[non_exhaustive]
pub enum InvalidBindAddressError {
#[error("cannot bind to requested address: {0}")]
CannotBindToRequestedAddress(ScionSocketIpAddr, Cow<'static, str>),
#[error(
"assigned address ({assigned_addr}) does not match requested address ({bind_addr}), likely due to NAT"
)]
AddressMismatch {
assigned_addr: ScionSocketIpAddr,
bind_addr: ScionSocketIpAddr,
},
#[error("could not find any local IP address to bind to")]
NoLocalIpAddressFound,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SnapConnectionError {
#[error("SNAP token source is missing")]
SnapTokenSourceMissing,
#[error("error establishing SNAP tunnel")]
TunnelEstablishment(#[source] Box<dyn std::error::Error + Send + Sync>),
#[error("failed to create SNAP control plane client")]
ControlPlaneClientCreation(#[source] Box<dyn std::error::Error + Send + Sync>),
#[error("failed to discover SNAP data plane")]
DataPlaneDiscovery(#[source] Box<dyn std::error::Error + Send + Sync>),
}
#[derive(Hash, Eq, PartialEq, Clone, Debug, Ord, PartialOrd)]
pub(crate) enum SocketKind {
Udp,
Scmp,
Raw,
}
pub(crate) trait DynUnderlayStack: Send + Sync {
fn bind_socket(
&self,
kind: SocketKind,
bind_addr: Option<ScionSocketIpAddr>,
) -> BoxFuture<'_, Result<BoundUnderlaySocket, ScionSocketBindError>>;
fn local_ases(&self) -> Vec<IsdAsn>;
}
pub(crate) struct BoundUnderlaySocket {
pub socket: Box<dyn UnderlaySocket>,
pub local_addr: ScionSocketIpAddr,
pub snap_data_plane: Option<net::SocketAddr>,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ScionSocketConnectError {
#[error("failed to get path to destination: {0}")]
PathLookupError(#[from] PathWaitTimeoutError),
#[error(transparent)]
BindError(#[from] ScionSocketBindError),
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ScionSocketSendError {
#[error("path lookup error: {0}")]
PathLookupError(#[from] PathWaitError),
#[error("udp next hop {address:?} unreachable: {isd_as}#{interface_id}: {msg}")]
UnderlayNextHopUnreachable {
isd_as: IsdAsn,
interface_id: u16,
address: Option<net::SocketAddr>,
msg: String,
},
#[error("invalid packet: {0}")]
InvalidPacket(Cow<'static, str>),
#[error("underlying socket is closed")]
Closed,
#[error("underlying connection returned an I/O error: {0:?}")]
IoError(#[from] std::io::Error),
#[error("socket is not connected")]
NotConnected,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ScionSocketReceiveError {
#[error("i/o error: {0:?}")]
IoError(#[from] std::io::Error),
#[error("socket is not connected")]
NotConnected,
}
pub(crate) const MAX_UNDERLAY_PACKET_SIZE: usize = 65535;
#[async_trait]
pub(crate) trait UnderlaySocket: 'static + Send + Sync {
fn try_send(&self, packet: &ScionRawPacketView) -> Result<(), ScionSocketSendError>;
async fn writeable(&self);
fn try_recv(&self, buf: &mut [u8]) -> Result<usize, ScionSocketReceiveError>;
async fn readable(&self);
}
#[async_trait]
pub(crate) trait UnderlaySocketExt: UnderlaySocket {
async fn send(&self, packet: &ScionRawPacketView) -> Result<(), ScionSocketSendError>;
async fn recv(&self, buf: &mut [u8]) -> Result<usize, ScionSocketReceiveError>;
}
#[async_trait]
impl<T: UnderlaySocket + ?Sized> UnderlaySocketExt for T {
async fn send(&self, packet: &ScionRawPacketView) -> Result<(), ScionSocketSendError> {
loop {
match self.try_send(packet) {
Err(ScionSocketSendError::IoError(e))
if e.kind() == std::io::ErrorKind::WouldBlock =>
{
self.writeable().await;
}
result => return result,
}
}
}
async fn recv(&self, buf: &mut [u8]) -> Result<usize, ScionSocketReceiveError> {
loop {
match self.try_recv(buf) {
Err(ScionSocketReceiveError::IoError(e))
if e.kind() == std::io::ErrorKind::WouldBlock =>
{
self.readable().await;
}
result => return result,
}
}
}
}
impl Drop for ScionStack {
fn drop(&mut self) {
tracing::warn!("ScionStack was dropped");
}
}