1mod priority_connect;
17
18use std::{borrow::Cow, fmt, net, sync::Arc, time::Duration};
19
20use endhost_api_client::client::CrpcEndhostApiClient;
21use rand::seq::IndexedRandom;
22use reqwest_connect_rpc::{
23 client::{CrpcClientCreationError, CrpcClientError},
24 token_source::{TokenSource, static_token::StaticTokenSource},
25};
26use scion_sdk_utils::backoff::ExponentialBackoff;
27use url::Url;
28use x25519_dalek::StaticSecret;
29
30pub use crate::underlays::udp::{OutboundIpResolver, TargetAddrOutboundIpResolver};
31use crate::{
32 ea_source::{
33 EndhostApiSource, EndhostApiSourceError, StaticEndhostApiDiscovery, StaticEndhostApis,
34 },
35 path::fetcher::{EndhostApiSegmentFetcher, traits::SegmentFetcher},
36 stack::ScionStack,
37 underlays::{
38 SnapSocketConfig, UnderlayStack,
39 discovery::{PeriodicUnderlayDiscovery, UnderlayDiscovery},
40 },
41};
42
43const DEFAULT_UDP_NEXT_HOP_RESOLVER_FETCH_INTERVAL: Duration = Duration::from_secs(600);
44const DEFAULT_ENDHOST_API_DISCOVERY_MAX_GROUPS: usize = 5;
45const DEFAULT_ENDHOST_API_DISCOVERY_APIS_PER_GROUP: usize = 2;
46const DEFAULT_ENDHOST_API_DISCOVERY_PER_GROUP_DELAY: Duration = Duration::from_millis(500);
47
48type OutboundIpResolverFactory = Box<dyn FnOnce(Url) -> Arc<dyn OutboundIpResolver> + Send>;
50
51pub struct ScionStackBuilder {
71 crpc_client: Option<reqwest::Client>,
72 endhost_api_token_source: Option<Arc<dyn TokenSource>>,
73 auth_token_source: Option<Arc<dyn TokenSource>>,
74 endhost_api_source: Arc<dyn EndhostApiSource>,
75 preferred_underlay: PreferredUnderlay,
76 endhost_api_discovery: EndhostApiDiscoveryConfig,
77 snap: SnapUnderlayConfig,
78 udp: UdpUnderlayConfig,
79}
80
81impl ScionStackBuilder {
82 #[must_use]
87 pub fn new() -> Self {
88 Self {
89 crpc_client: None,
90 endhost_api_token_source: None,
91 auth_token_source: None,
92 endhost_api_source: Arc::new(StaticEndhostApiDiscovery::global()),
93 preferred_underlay: PreferredUnderlay::Udp,
94 endhost_api_discovery: EndhostApiDiscoveryConfig::default(),
95 snap: SnapUnderlayConfig::default(),
96 udp: UdpUnderlayConfig::default(),
97 }
98 }
99
100 #[must_use]
104 pub fn with_preferred_underlay(mut self, preferred: PreferredUnderlay) -> Self {
105 self.preferred_underlay = preferred;
106 self
107 }
108
109 #[must_use]
114 pub fn with_crpc_client(mut self, crpc_client: reqwest::Client) -> Self {
115 self.crpc_client = Some(crpc_client);
116 self
117 }
118
119 #[must_use]
125 pub fn with_endhost_api(mut self, endhost_api_url: Url) -> Self {
126 let source = StaticEndhostApis::new().add_group(vec![endhost_api_url]);
127 self.endhost_api_source = Arc::new(source);
128
129 self
130 }
131
132 #[must_use]
136 pub fn with_endhost_api_discovery_source(mut self, source: impl EndhostApiSource) -> Self {
137 self.endhost_api_source = Arc::new(source);
138 self
139 }
140
141 #[must_use]
143 pub fn with_endhost_api_auth_token_source(mut self, source: impl TokenSource) -> Self {
144 self.endhost_api_token_source = Some(Arc::new(source));
145 self
146 }
147
148 #[must_use]
150 pub fn with_endhost_api_auth_token(mut self, token: String) -> Self {
151 self.endhost_api_token_source = Some(Arc::new(StaticTokenSource::from(token)));
152 self
153 }
154
155 #[must_use]
159 pub fn with_auth_token_source(mut self, source: impl TokenSource) -> Self {
160 self.auth_token_source = Some(Arc::new(source));
161 self
162 }
163
164 #[must_use]
168 pub fn with_auth_token(mut self, token: String) -> Self {
169 self.auth_token_source = Some(Arc::new(StaticTokenSource::from(token)));
170 self
171 }
172
173 #[must_use]
179 pub fn with_endhost_api_discovery_max_groups(mut self, max_groups: usize) -> Self {
180 self.endhost_api_discovery.max_groups = max_groups;
181 self
182 }
183
184 #[must_use]
191 pub fn with_anapaya_ead_apis_per_group(mut self, apis_per_group: usize) -> Self {
192 self.endhost_api_discovery.apis_per_group = apis_per_group;
193 self
194 }
195
196 #[must_use]
205 pub fn with_endhost_api_discovery_per_group_delay(mut self, per_group_delay: Duration) -> Self {
206 self.endhost_api_discovery.per_group_delay = per_group_delay;
207 self
208 }
209
210 #[must_use]
212 pub fn with_snap_underlay_config(mut self, config: SnapUnderlayConfig) -> Self {
213 self.snap = config;
214 self
215 }
216
217 #[must_use]
219 pub fn with_udp_underlay_config(mut self, config: UdpUnderlayConfig) -> Self {
220 self.udp = config;
221 self
222 }
223
224 pub async fn build(self) -> Result<ScionStack, BuildScionStackError> {
230 let ScionStackBuilder {
231 crpc_client,
232 endhost_api_token_source,
233 auth_token_source,
234 endhost_api_source,
235 preferred_underlay,
236 endhost_api_discovery,
237 snap,
238 udp,
239 } = self;
240
241 let api_groups = endhost_api_source.endhost_apis().await?;
246 let api_groups: Vec<Vec<Url>> = {
247 let mut rng = rand::rng();
248 api_groups
249 .into_iter()
250 .map(|g| g.apis.into_iter().map(|a| a.address).collect::<Vec<_>>())
251 .filter(|group| !group.is_empty())
252 .take(endhost_api_discovery.max_groups)
253 .map(|group: Vec<Url>| {
254 group
255 .sample(&mut rng, endhost_api_discovery.apis_per_group)
256 .cloned()
257 .collect()
258 })
259 .collect()
260 };
261
262 if api_groups.is_empty() {
263 return Err(BuildScionStackError::EndhostApiSourceError(
265 EndhostApiSourceError::new("endhost API discovery returned no APIs", false),
266 ));
267 }
268
269 let token_source: Option<Arc<dyn TokenSource>> =
270 endhost_api_token_source.or(auth_token_source.clone());
271 let crpc_c = crpc_client.clone();
272 let discover_underlays = move |url: Url| {
273 let token_source = token_source.clone();
274 let crpc_c = crpc_c.clone();
275 let url = url.clone();
276 async move {
277 let mut client = match crpc_c {
278 Some(client) => {
279 CrpcEndhostApiClient::new_with_client(&url, client)
280 .map_err(ApiAttemptError::ClientSetup)?
281 }
282 None => {
283 CrpcEndhostApiClient::new(&url).map_err(ApiAttemptError::ClientSetup)?
284 }
285 };
286 if let Some(token_source) = &token_source {
287 client.use_token_source(token_source.clone());
288 }
289 let client = Arc::new(client);
290 let discovery = PeriodicUnderlayDiscovery::new(
291 client.clone(),
292 udp.udp_next_hop_resolver_fetch_interval,
293 ExponentialBackoff::new(0.5, 10.0, 2.0, 0.5),
294 )
295 .await
296 .map_err(ApiAttemptError::UnderlayDiscovery)?;
297 Ok((client, discovery))
298 }
299 };
300
301 let (api_url, (endhost_api_client, underlay_discovery)) =
302 priority_connect::try_priority_groups(
303 api_groups,
304 discover_underlays,
305 endhost_api_discovery.per_group_delay,
306 )
307 .await
308 .map_err(|errors| {
309 BuildScionStackError::AllEndhostApisFailed(AllEndhostApisFailed::new(errors))
310 })?;
311 tracing::info!(url=%api_url, "Selected endhost API");
312
313 let outbound_ip_resolver: Arc<dyn OutboundIpResolver> =
317 (udp.outbound_ip_resolver_factory)(api_url.clone());
318
319 let underlay_stack = UnderlayStack::new(
320 preferred_underlay,
321 Arc::new(underlay_discovery),
322 outbound_ip_resolver,
323 snap.static_identity.unwrap_or_else(StaticSecret::random),
324 SnapSocketConfig {
325 crpc_client: snap.crpc_client.or(crpc_client),
326 snap_token_source: snap.snap_token_source.or(auth_token_source),
327 },
328 );
329
330 Ok(ScionStack::new(
331 Some(api_url),
332 Arc::new(EndhostApiSegmentFetcher::new(endhost_api_client)),
333 Arc::new(underlay_stack),
334 ))
335 }
336
337 fn build_static_udp_underlay(
352 underlay_discovery: Arc<dyn UnderlayDiscovery>,
353 outbound_ip_resolver: Arc<dyn OutboundIpResolver>,
354 default_segment_fetcher: Arc<dyn SegmentFetcher>,
355 ) -> ScionStack {
356 let underlay_stack = UnderlayStack::new(
357 PreferredUnderlay::Udp,
358 underlay_discovery,
359 outbound_ip_resolver,
360 StaticSecret::random(),
361 SnapSocketConfig {
362 crpc_client: None,
363 snap_token_source: None,
364 },
365 );
366
367 ScionStack::new(None, default_segment_fetcher, Arc::new(underlay_stack))
368 }
369}
370
371impl Default for ScionStackBuilder {
372 fn default() -> Self {
373 Self::new()
374 }
375}
376
377impl ScionStack {
378 #[must_use]
393 pub fn static_udp_underlay(
394 underlay_discovery: Arc<dyn UnderlayDiscovery>,
395 outbound_ip_resolver: Arc<dyn OutboundIpResolver>,
396 default_segment_fetcher: Arc<dyn SegmentFetcher>,
397 ) -> ScionStack {
398 ScionStackBuilder::build_static_udp_underlay(
399 underlay_discovery,
400 outbound_ip_resolver,
401 default_segment_fetcher,
402 )
403 }
404}
405
406#[derive(thiserror::Error, Debug)]
408#[non_exhaustive]
409pub enum BuildScionStackError {
410 #[error("no underlay available: {0}")]
412 UnderlayUnavailable(Cow<'static, str>),
413 #[error(transparent)]
415 AllEndhostApisFailed(#[from] AllEndhostApisFailed),
416 #[error(transparent)]
418 EndhostApiSourceError(#[from] EndhostApiSourceError),
419 #[error(transparent)]
422 Snap(#[from] BuildSnapScionStackError),
423 #[error("internal error")]
425 Internal(#[source] Box<dyn std::error::Error + Send + Sync>),
426}
427
428impl BuildScionStackError {
429 #[must_use]
434 pub fn is_transient(&self) -> bool {
435 match self {
436 Self::UnderlayUnavailable(_) => false,
438 Self::AllEndhostApisFailed(failed) => failed.is_transient(),
439 Self::EndhostApiSourceError(source) => source.is_transient(),
440 Self::Snap(error) => error.is_transient(),
441 Self::Internal(_) => false,
443 }
444 }
445}
446
447#[derive(thiserror::Error, Debug)]
452#[non_exhaustive]
453pub enum BuildSnapScionStackError {
454 #[error("no SNAP data plane available: {0}")]
456 DataPlaneUnavailable(Cow<'static, str>),
457 #[error("control plane client setup error")]
459 ControlPlaneClientSetup(#[source] Box<dyn std::error::Error + Send + Sync>),
460 #[error("data plane discovery request error")]
462 DataPlaneDiscovery(#[source] Box<dyn std::error::Error + Send + Sync>),
463}
464
465impl BuildSnapScionStackError {
466 #[must_use]
471 pub fn is_transient(&self) -> bool {
472 match self {
473 Self::DataPlaneUnavailable(_) => false,
475 Self::ControlPlaneClientSetup(_) | Self::DataPlaneDiscovery(_) => true,
477 }
478 }
479}
480
481#[derive(Debug)]
485pub struct AllEndhostApisFailed {
486 failures: Vec<(Url, ApiAttemptError)>,
487}
488
489impl AllEndhostApisFailed {
490 pub(crate) fn new(failures: Vec<(Url, ApiAttemptError)>) -> Self {
491 Self { failures }
492 }
493
494 #[must_use]
496 pub fn failures(&self) -> &[(Url, ApiAttemptError)] {
497 &self.failures
498 }
499
500 #[must_use]
503 pub fn is_transient(&self) -> bool {
504 !self.failures.is_empty() && self.failures.iter().all(|(_, err)| err.is_transient())
505 }
506}
507
508impl fmt::Display for AllEndhostApisFailed {
509 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
510 write!(f, "all {} endhost API(s) failed", self.failures.len())?;
511 let mut sep = ": ";
512 for (url, err) in &self.failures {
513 write!(f, "{sep}{url} ({err})")?;
514 sep = "; ";
515 }
516 Ok(())
517 }
518}
519
520impl std::error::Error for AllEndhostApisFailed {}
521
522#[derive(thiserror::Error, Debug)]
526#[non_exhaustive]
527pub enum ApiAttemptError {
528 #[error("client setup")]
530 ClientSetup(#[source] CrpcClientCreationError),
531 #[error("underlay discovery")]
533 UnderlayDiscovery(#[source] CrpcClientError),
534}
535
536impl ApiAttemptError {
537 #[must_use]
542 pub fn is_transient(&self) -> bool {
543 match self {
544 Self::ClientSetup(error) => error.is_transient(),
545 Self::UnderlayDiscovery(error) => error.is_transient(),
546 }
547 }
548}
549
550pub struct EndhostApiDiscoveryConfig {
555 max_groups: usize,
557 apis_per_group: usize,
559 per_group_delay: Duration,
562}
563
564impl Default for EndhostApiDiscoveryConfig {
565 fn default() -> Self {
566 Self {
567 max_groups: DEFAULT_ENDHOST_API_DISCOVERY_MAX_GROUPS,
568 apis_per_group: DEFAULT_ENDHOST_API_DISCOVERY_APIS_PER_GROUP,
569 per_group_delay: DEFAULT_ENDHOST_API_DISCOVERY_PER_GROUP_DELAY,
570 }
571 }
572}
573
574#[derive(Debug, Clone, Copy, PartialEq, Eq)]
576#[non_exhaustive]
577pub enum PreferredUnderlay {
578 Snap,
580 Udp,
582}
583
584#[derive(Default)]
589pub struct SnapUnderlayConfig {
590 crpc_client: Option<reqwest::Client>,
591 snap_token_source: Option<Arc<dyn TokenSource>>,
592 snap_dp_index: usize,
593 static_identity: Option<StaticSecret>,
595}
596
597impl SnapUnderlayConfig {
598 #[must_use]
600 pub fn with_auth_token(mut self, token: String) -> Self {
601 self.snap_token_source = Some(Arc::new(StaticTokenSource::from(token)));
602 self
603 }
604
605 #[must_use]
607 pub fn with_auth_token_source(mut self, source: impl TokenSource) -> Self {
608 self.snap_token_source = Some(Arc::new(source));
609 self
610 }
611
612 #[must_use]
614 pub fn with_crpc_client(mut self, client: reqwest::Client) -> Self {
615 self.crpc_client = Some(client);
616 self
617 }
618
619 #[must_use]
621 pub fn with_snap_dp_index(mut self, dp_index: usize) -> Self {
622 self.snap_dp_index = dp_index;
623 self
624 }
625
626 #[must_use]
630 pub fn with_static_identity(mut self, identity: StaticSecret) -> Self {
631 self.static_identity = Some(identity);
632 self
633 }
634}
635
636pub struct UdpUnderlayConfig {
641 udp_next_hop_resolver_fetch_interval: Duration,
642 outbound_ip_resolver_factory: OutboundIpResolverFactory,
643}
644
645impl Default for UdpUnderlayConfig {
646 fn default() -> Self {
647 Self {
648 udp_next_hop_resolver_fetch_interval: DEFAULT_UDP_NEXT_HOP_RESOLVER_FETCH_INTERVAL,
649 outbound_ip_resolver_factory: Box::new(move |url| {
650 Arc::new(TargetAddrOutboundIpResolver::new(url, vec![]))
651 }),
652 }
653 }
654}
655
656impl UdpUnderlayConfig {
657 #[must_use]
663 pub fn with_outbound_ips(mut self, outbound_ips: Vec<net::IpAddr>) -> Self {
664 self.outbound_ip_resolver_factory =
665 Box::new(move |_url| Arc::new(outbound_ips) as Arc<dyn OutboundIpResolver>);
666 self
667 }
668
669 #[must_use]
678 pub fn with_outbound_ip_resolver(
679 mut self,
680 resolver: impl OutboundIpResolver + 'static,
681 ) -> Self {
682 let resolver = Arc::new(resolver) as Arc<dyn OutboundIpResolver>;
683 self.outbound_ip_resolver_factory = Box::new(move |_url| resolver.clone());
684 self
685 }
686
687 #[must_use]
699 pub fn with_outbound_ip_resolver_factory<F, R>(mut self, factory: F) -> Self
700 where
701 F: FnOnce(Url) -> R + Send + 'static,
702 R: OutboundIpResolver + 'static,
703 {
704 self.outbound_ip_resolver_factory =
705 Box::new(move |url| Arc::new(factory(url)) as Arc<dyn OutboundIpResolver>);
706 self
707 }
708
709 #[must_use]
712 pub fn with_udp_next_hop_resolver_fetch_interval(mut self, fetch_interval: Duration) -> Self {
713 self.udp_next_hop_resolver_fetch_interval = fetch_interval;
714 self
715 }
716}
717
718#[cfg(test)]
719mod tests {
720 use std::borrow::Cow;
721
722 use reqwest::header;
723 use reqwest_connect_rpc::client::CrpcClientError;
724 use url::Url;
725
726 use super::*;
727
728 fn connection_error() -> CrpcClientError {
729 CrpcClientError::ConnectionError {
730 context: Cow::Borrowed("test"),
731 source: Box::new(std::io::Error::other("boom")),
732 }
733 }
734
735 fn non_connection_error() -> CrpcClientError {
736 CrpcClientError::DecodeError {
737 context: Cow::Borrowed("test"),
738 source: Some(Box::new(std::io::Error::other("boom"))),
739 body: None,
740 }
741 }
742
743 fn client_creation_error() -> CrpcClientCreationError {
744 CrpcClientCreationError::InvalidUserAgent {
745 user_agent: "in\nvalid".to_owned(),
746 source: header::HeaderValue::from_str("in\nvalid").expect_err("invalid header value"),
747 }
748 }
749
750 #[test]
751 fn api_attempt_error_transient_classification() {
752 assert!(ApiAttemptError::UnderlayDiscovery(connection_error()).is_transient());
754 assert!(!ApiAttemptError::UnderlayDiscovery(non_connection_error()).is_transient());
756 assert!(!ApiAttemptError::ClientSetup(client_creation_error()).is_transient());
758 }
759
760 #[test]
761 fn all_endhost_apis_failed_transient_classification() {
762 let url: Url = "http://example.com".parse().expect("valid url");
763
764 assert!(!AllEndhostApisFailed::new(vec![]).is_transient());
766
767 assert!(
769 AllEndhostApisFailed::new(vec![(
770 url.clone(),
771 ApiAttemptError::UnderlayDiscovery(connection_error()),
772 )])
773 .is_transient()
774 );
775
776 assert!(
778 !AllEndhostApisFailed::new(vec![
779 (
780 url.clone(),
781 ApiAttemptError::UnderlayDiscovery(connection_error()),
782 ),
783 (
784 url,
785 ApiAttemptError::UnderlayDiscovery(non_connection_error())
786 ),
787 ])
788 .is_transient()
789 );
790 }
791}