From 23dc906da0fc2e7045c31ce7edbb16ecfe1ce782 Mon Sep 17 00:00:00 2001 From: Ben_Kosytorz Date: Thu, 24 Sep 2026 15:58:12 +0200 Subject: [PATCH] sn-fetch: per-exit egress (Direct vs Proxy) for neighbor exits over the mesh --- server/crates/sn-fetch/src/exit.rs | 31 +++++ server/crates/sn-fetch/src/main.rs | 37 ++++-- server/crates/sn-fetch/src/plan.rs | 15 ++- server/crates/sn-fetch/src/runner.rs | 38 ++++-- server/crates/sn-fetch/src/types.rs | 33 +++++ server/crates/sn-fetch/tests/mesh_exit.rs | 151 ++++++++++++++++++++++ server/crates/sn-fetch/tests/roundtrip.rs | 12 +- 7 files changed, 288 insertions(+), 29 deletions(-) create mode 100644 server/crates/sn-fetch/tests/mesh_exit.rs diff --git a/server/crates/sn-fetch/src/exit.rs b/server/crates/sn-fetch/src/exit.rs index 92ebe14..fa2a13e 100644 --- a/server/crates/sn-fetch/src/exit.rs +++ b/server/crates/sn-fetch/src/exit.rs @@ -48,6 +48,37 @@ impl HttpExit for LocalExit { } } +/// A neighbor's uplink, reached through an HTTP proxy over the mesh. +/// The proxy is expected to listen on the neighbor's Yggdrasil address, +/// so the transport between the households is encrypted end-to-end by +/// the overlay. +#[derive(Debug)] +pub struct MeshExit { + id: ExitId, + client: reqwest::Client, +} + +impl MeshExit { + pub fn new(id: ExitId, proxy: reqwest::Url) -> Result { + let proxy = reqwest::Proxy::all(proxy)?; + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(30)) + .proxy(proxy) + .build()?; + Ok(Self { id, client }) + } +} + +impl HttpExit for MeshExit { + fn id(&self) -> ExitId { + self.id + } + + fn client(&self) -> &reqwest::Client { + &self.client + } +} + /// All exits available for a fetch, keyed by id. #[derive(Default, Clone)] pub struct ExitRegistry { diff --git a/server/crates/sn-fetch/src/main.rs b/server/crates/sn-fetch/src/main.rs index be24782..c78e517 100644 --- a/server/crates/sn-fetch/src/main.rs +++ b/server/crates/sn-fetch/src/main.rs @@ -15,7 +15,7 @@ use std::path::PathBuf; use clap::Parser; use sn_fetch::runner::{self, FetchOptions}; use sn_fetch::types::{ - ExitId, ExitWeight, MinSegmentBytes, SegmentsPerExit, Sha256Digest, TargetUrl, + Egress, ExitId, ExitWeight, MinSegmentBytes, SegmentsPerExit, Sha256Digest, TargetUrl, }; #[derive(Debug, Parser)] @@ -28,12 +28,20 @@ struct Args { #[arg(short, long)] output: Option, - /// Exits as `id=weight` pairs, comma separated, e.g. `local=1,local2=2`. - /// `PoC`: all ids map to plain local connections; the weights only shape - /// the plan. Multi-exit routing comes with the mesh integration. - #[arg(short = 'e', long = "exit", default_value = "local=1")] + /// Exits as `id=weight` pairs, comma separated, e.g. `1=2,2=1`. + /// A proxy route turns an exit into a neighbor's uplink over the + /// mesh overlay: `id=weight@http://[ygg-addr]:8080`. Without a + /// proxy, the exit is this machine's own connection. + /// `PoC`: weights only shape the plan; a real proxy service + /// (`sn-exit`) comes with the mesh integration. + #[arg(short = 'e', long = "exit", default_value = "1=1")] exits: String, + /// Total tries per segment (initial try + retries; failed segments + /// are retried on a different exit when available). + #[arg(long, default_value_t = 3)] + max_attempts: u32, + /// Parallel sub-segments per exit. #[arg(short = 's', long, default_value_t = 1)] segments_per_exit: usize, @@ -51,10 +59,10 @@ struct Args { probe_only: bool, } -fn parse_exits(spec: &str) -> Result, String> { +fn parse_exits(spec: &str) -> Result, String> { spec.split(',') .map(|pair| { - let (id, weight) = pair + let (id, rest) = pair .split_once('=') .ok_or_else(|| format!("invalid exit spec `{pair}`, expected `id=weight`"))?; let id = id @@ -63,13 +71,21 @@ fn parse_exits(spec: &str) -> Result, String> { .ok() .and_then(NonZeroU8::new) .ok_or_else(|| format!("invalid exit id `{id}` (must be a non-zero number)"))?; + let (weight, egress) = match rest.split_once('@') { + Some((weight, proxy)) => { + let egress = Egress::proxy(proxy.trim()) + .map_err(|err| format!("invalid proxy in exit spec `{pair}`: {err}"))?; + (weight, egress) + } + None => (rest, Egress::Direct), + }; let weight = weight .trim() .parse::() .ok() .and_then(NonZeroU32::new) .ok_or_else(|| format!("invalid exit weight `{weight}` (must be non-zero)"))?; - Ok((ExitId::new(id), ExitWeight::new(weight))) + Ok((ExitId::new(id), ExitWeight::new(weight), egress)) }) .collect() } @@ -131,6 +147,8 @@ async fn run() -> Result<(), String> { NonZeroU64::new(args.min_segment_mib.saturating_mul(1024 * 1024)) .ok_or("min_segment_mib must be at least 1")?, ); + let max_attempts = + NonZeroU32::new(args.max_attempts).ok_or("max_attempts must be at least 1")?; let expected = match &args.sha256 { Some(hex) => Some(Sha256Digest::from_hex(hex).map_err(|err| err.to_string())?), @@ -146,9 +164,10 @@ async fn run() -> Result<(), String> { segments_per_exit: SegmentsPerExit::new(spe), min_segment, expected_sha256: expected, + retries: sn_fetch::exec::RetryPolicy::new(max_attempts), }; - let registry = runner::local_registry(&exit_specs).map_err(|err| err.to_string())?; + let registry = runner::registry_from_specs(&exit_specs).map_err(|err| err.to_string())?; let parts_dir = std::env::temp_dir().join("sn-fetch-parts"); let report = runner::run(&target, &options, ®istry, &parts_dir) diff --git a/server/crates/sn-fetch/src/plan.rs b/server/crates/sn-fetch/src/plan.rs index f5ffc08..e2dd066 100644 --- a/server/crates/sn-fetch/src/plan.rs +++ b/server/crates/sn-fetch/src/plan.rs @@ -2,8 +2,8 @@ use crate::error::FetchError; use crate::types::{ - ByteRange, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentIndex, SegmentsPerExit, - TargetUrl, + ByteRange, Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentIndex, + SegmentsPerExit, TargetUrl, }; use std::num::NonZeroUsize; @@ -23,8 +23,9 @@ pub struct FetchPlan { pub segments: Vec, } -/// Exits with their capacity weights. -pub type ExitSpecs = Vec<(ExitId, ExitWeight)>; +/// exit description, including how traffic leaves (`Direct` = own +/// uplink, `Proxy` = a neighbor's exit over the mesh). +pub type ExitSpecs = Vec<(ExitId, ExitWeight, Egress)>; impl FetchPlan { /// Build a validated plan. @@ -56,12 +57,12 @@ impl FetchPlan { let weakest = exits .iter() .enumerate() - .min_by_key(|(_, (id, weight))| (weight.value(), std::cmp::Reverse(id.value()))) + .min_by_key(|(_, (id, weight, _))| (weight.value(), std::cmp::Reverse(id.value()))) .map_or(0, |(pos, _)| pos); exits.remove(weakest); } // Deterministic boundary assignment order. - exits.sort_by_key(|(id, _)| *id); + exits.sort_by_key(|(id, _, _)| *id); // If even one segment per exit would be too small, reduce to one // segment per exit. @@ -85,7 +86,7 @@ impl FetchPlan { // Weighted split across exits, then even sub-split per exit. let units: Vec = exits .iter() - .map(|(_, w)| u128::from(w.value()) * spe.value() as u128) + .map(|(_, w, _)| u128::from(w.value()) * spe.value() as u128) .collect(); let total_units: u128 = units.iter().sum(); diff --git a/server/crates/sn-fetch/src/runner.rs b/server/crates/sn-fetch/src/runner.rs index 3e2eb2d..504f7a9 100644 --- a/server/crates/sn-fetch/src/runner.rs +++ b/server/crates/sn-fetch/src/runner.rs @@ -10,12 +10,13 @@ use sha2::Digest; use tokio::io::AsyncWriteExt; use crate::error::FetchError; -use crate::exec::{self, FetchReport, SegmentStats}; -use crate::exit::{ExitRegistry, LocalExit}; +use crate::exec::{self, FetchReport, RetryPolicy, SegmentStats}; +use crate::exit::{ExitRegistry, HttpExit, LocalExit, MeshExit}; use crate::plan::ExitSpecs; use crate::probe::{self, ProbeInfo}; +use crate::progress::ProgressSink; use crate::types::{ - FileSize, MinSegmentBytes, RangeSupport, SegmentsPerExit, Sha256Digest, TargetUrl, + Egress, FileSize, MinSegmentBytes, RangeSupport, SegmentsPerExit, Sha256Digest, TargetUrl, }; /// All knobs for one fetch. @@ -26,6 +27,8 @@ pub struct FetchOptions { pub segments_per_exit: SegmentsPerExit, pub min_segment: MinSegmentBytes, pub expected_sha256: Option, + /// How often a segment is retried (preferably on another exit). + pub retries: RetryPolicy, } /// Run a full fetch and produce a report. @@ -74,7 +77,17 @@ async fn run_ranged( tracing::info!("plan: {} segment(s)", plan.segments.len()); let started = Instant::now(); - let segment_stats: Vec = exec::run_segments(&plan, exits, parts_dir).await?; + let segment_stats: Vec = exec::run_segments_with_policy( + &plan, + exits, + parts_dir, + &options.retries, + &ProgressSink::tracing_log(), + ) + .await? + .into_iter() + .map(std::convert::From::from) + .collect(); let digest = exec::assemble( parts_dir, &options.output, @@ -105,7 +118,7 @@ async fn run_whole( let exit_id = options .exits .first() - .map(|(id, _)| *id) + .map(|(id, _, _)| *id) .ok_or_else(|| FetchError::InvalidPlan("no exits given".to_string()))?; let client = exits.get(exit_id)?.client().clone(); @@ -163,11 +176,18 @@ async fn run_whole( }) } -/// Convenience: build a registry with one [`LocalExit`] per spec id. -pub fn local_registry(specs: &ExitSpecs) -> Result { +/// Build a registry from exit specs: `Egress::Direct` exits connect +/// locally (own uplink), `Egress::Proxy` exits route through the given +/// proxy — typically a neighbor's exit service over the Yggdrasil +/// overlay. +pub fn registry_from_specs(specs: &ExitSpecs) -> Result { let mut registry = ExitRegistry::empty(); - for (id, _) in specs { - registry.register(std::sync::Arc::new(LocalExit::new(*id)?)); + for (id, _, egress) in specs { + let exit: std::sync::Arc = match egress { + Egress::Direct => std::sync::Arc::new(LocalExit::new(*id)?), + Egress::Proxy(url) => std::sync::Arc::new(MeshExit::new(*id, url.clone())?), + }; + registry.register(exit); } Ok(registry) } diff --git a/server/crates/sn-fetch/src/types.rs b/server/crates/sn-fetch/src/types.rs index b1803b9..71144ef 100644 --- a/server/crates/sn-fetch/src/types.rs +++ b/server/crates/sn-fetch/src/types.rs @@ -230,3 +230,36 @@ pub enum RangeSupport { /// Probe was inconclusive; treat as unsupported (safe fallback). Unknown, } + +/// How a request physically leaves through an exit. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Egress { + /// The household's own uplink: plain outgoing HTTP, no detour. + Direct, + /// Through an HTTP proxy — e.g. a neighbor's exit service reachable + /// over the mesh overlay (typically bound to their Yggdrasil + /// address). + Proxy(reqwest::Url), +} + +impl Egress { + /// Parse a proxy URL for [`Egress::Proxy`]; only `http`/`https` + /// upstreams are accepted. + pub fn proxy(input: &str) -> Result { + let url = + reqwest::Url::parse(input).map_err(|_| FetchError::InvalidUrl(input.to_string()))?; + match url.scheme() { + "http" | "https" => Ok(Self::Proxy(url)), + _ => Err(FetchError::InvalidUrl(input.to_string())), + } + } +} + +impl fmt::Display for Egress { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Direct => write!(f, "direct"), + Self::Proxy(url) => write!(f, "proxy:{url}"), + } + } +} diff --git a/server/crates/sn-fetch/tests/mesh_exit.rs b/server/crates/sn-fetch/tests/mesh_exit.rs new file mode 100644 index 0000000..941f7d2 --- /dev/null +++ b/server/crates/sn-fetch/tests/mesh_exit.rs @@ -0,0 +1,151 @@ +//! End-to-end test: a `MeshExit` fetches through a real HTTP proxy. +//! +//! The fake proxy is a minimal HTTP forward proxy: it accepts absolute-URI +//! requests (`GET http://origin/file.bin HTTP/1.1`), extracts the `Range` +//! header and serves the requested slice from an in-memory buffer — +//! exactly the shape of an `sn-exit` service bound to a neighbor's +//! Yggdrasil address (minus the overlay transport, which is transparent +//! to HTTP). + +#![forbid(unsafe_code)] +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::num::{NonZeroU32, NonZeroU64, NonZeroU8, NonZeroUsize}; +use std::sync::Arc; + +use sha2::{Digest, Sha256}; +use tempfile::TempDir; + +use sn_fetch::exit::{ExitRegistry, MeshExit}; +use sn_fetch::plan::{ExitSpecs, FetchPlan}; +use sn_fetch::types::{ + Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentsPerExit, TargetUrl, +}; + +/// Minimal forward proxy serving one fixed object from memory. +async fn spawn_proxy(data: Arc>) -> std::net::SocketAddr { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + break; + }; + let data = Arc::clone(&data); + tokio::spawn(async move { + // Read the request head. + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.windows(4).any(|w| w == b"\r\n\r\n") { + if socket.read(&mut byte).await.unwrap_or(0) == 0 { + return; + } + head.push(byte[0]); + } + let head_text = String::from_utf8_lossy(&head); + + // First line: `GET http://origin/path HTTP/1.1`. + let first_line = head_text.lines().next().unwrap_or_default(); + if !first_line.starts_with("GET http://") { + socket + .write_all(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n") + .await + .ok(); + return; + } + + // Extract the requested byte range (if any). Header + // names arrive lowercased from some clients. + let range = head_text.lines().find_map(|line| { + let (name, value) = line.split_once(':')?; + if !name.eq_ignore_ascii_case("range") { + return None; + } + let value = value.trim().strip_prefix("bytes=")?; + let (start, end) = value.trim().split_once('-')?; + Some((start.parse::().ok()?, end.parse::().ok()?)) + }); + + let (start, end_inclusive, status) = match range { + Some((start, end)) => (start, end, "206 Partial Content"), + None => (0, data.len() as u64 - 1, "200 OK"), + }; + let slice = &data[start as usize..=(end_inclusive as usize).min(data.len() - 1)]; + let response = format!( + "HTTP/1.1 {status}\r\nContent-Length: {}\r\n\r\n", + slice.len() + ); + socket.write_all(response.as_bytes()).await.ok(); + socket.write_all(slice).await.ok(); + socket.shutdown().await.ok(); + }); + } + }); + addr +} + +#[tokio::test] +async fn mesh_exit_downloads_through_the_proxy() { + let data: Vec = (0..64_000u32).map(|i| (i % 251) as u8).collect(); + let data = Arc::new(data.clone()); + let proxy_addr = spawn_proxy(Arc::clone(&data)).await; + + let proxy_url = format!("http://{proxy_addr}"); + let specs: ExitSpecs = vec![( + ExitId::new(NonZeroU8::new(1).expect("1 is non-zero")), + ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")), + Egress::proxy(&proxy_url).expect("proxy url is valid"), + )]; + + // The origin URL is fictional — the proxy serves the content; this + // proves the request really went through the proxy, not straight to + // an origin server. + let target = TargetUrl::parse("http://origin.invalid/file.bin").expect("valid url"); + let file_size = FileSize::try_new(data.len() as u64).expect("non-zero in test"); + + let plan = FetchPlan::build( + target.clone(), + file_size, + &specs, + SegmentsPerExit::new(NonZeroUsize::new(2).expect("2 is non-zero")), + MinSegmentBytes::new(NonZeroU64::new(1024).expect("1024 is non-zero")), + ) + .expect("plan builds"); + plan.validate().expect("plan validates"); + + let mut registry = ExitRegistry::empty(); + registry.register(Arc::new( + MeshExit::new( + ExitId::new(NonZeroU8::new(1).expect("1 is non-zero")), + reqwest::Url::parse(&proxy_url).expect("proxy url is valid"), + ) + .expect("mesh exit builds"), + )); + + let parts = TempDir::new().expect("tempdir in test"); + let stats = sn_fetch::exec::run_segments(&plan, ®istry, parts.path()) + .await + .expect("segments succeed through the proxy"); + let total: u64 = stats.iter().map(|s| s.bytes).sum(); + assert_eq!(total, data.len() as u64); + + let output = parts.path().join("mesh.bin"); + let expected_digest = + sn_fetch::types::Sha256Digest::from_bytes(Sha256::digest(data.as_ref().as_slice()).into()); + let digest = sn_fetch::exec::assemble(parts.path(), &output, file_size, Some(&expected_digest)) + .await + .expect("assembly verifies"); + assert_eq!(digest, expected_digest); +} + +/// The proxy's address (IPv4, arbitrary port) is not a valid proxy URL +/// unless it has an http scheme — Egress::proxy must reject junk. +#[test] +fn egress_rejects_non_http_proxy_schemes() { + assert!(Egress::proxy("ftp://proxy:2121").is_err()); + assert!(Egress::proxy("not a url").is_err()); + assert!(Egress::proxy("http://127.0.0.1:8080").is_ok()); + assert!(Egress::proxy("https://proxy.example:3128").is_ok()); +} diff --git a/server/crates/sn-fetch/tests/roundtrip.rs b/server/crates/sn-fetch/tests/roundtrip.rs index e629bcb..457c536 100644 --- a/server/crates/sn-fetch/tests/roundtrip.rs +++ b/server/crates/sn-fetch/tests/roundtrip.rs @@ -21,8 +21,8 @@ use sn_fetch::exit::LocalExit; use sn_fetch::plan::{ExitSpecs, FetchPlan}; use sn_fetch::probe; use sn_fetch::types::{ - ByteRange, ExitId, ExitWeight, FileSize, MinSegmentBytes, RangeSupport, SegmentsPerExit, - Sha256Digest, + ByteRange, Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, RangeSupport, + SegmentsPerExit, Sha256Digest, }; fn exit_specs(n: u8) -> ExitSpecs { @@ -31,6 +31,7 @@ fn exit_specs(n: u8) -> ExitSpecs { ( ExitId::new(NonZeroU8::new(i).expect("test ids are 1..=255")), ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")), + Egress::Direct, ) }) .collect() @@ -68,7 +69,7 @@ async fn segmented_download_reassembles_and_hashes() { plan.validate().expect("plan validates"); let mut registry = sn_fetch::exit::ExitRegistry::empty(); - for id in specs.iter().map(|(id, _)| *id) { + for id in specs.iter().map(|(id, _, _)| *id) { registry.register(Arc::new(LocalExit::new(id).expect("local exit"))); } @@ -109,8 +110,9 @@ async fn range_hostile_server_falls_back_to_single_stream() { segments_per_exit: SegmentsPerExit::new(NonZeroUsize::new(1).expect("1 is non-zero")), min_segment: MinSegmentBytes::new(NonZeroU64::new(1024).expect("1024 is non-zero")), expected_sha256: None, + retries: sn_fetch::exec::RetryPolicy::default(), }; - let registry = sn_fetch::runner::local_registry(&specs).expect("registry"); + let registry = sn_fetch::runner::registry_from_specs(&specs).expect("registry"); let parts = TempDir::new().expect("tempdir in test"); let report = sn_fetch::runner::run(&target, &options, ®istry, parts.path()) @@ -131,10 +133,12 @@ fn plan_is_contiguous_and_weighted() { ( ExitId::new(NonZeroU8::new(1).expect("1 is non-zero")), ExitWeight::new(NonZeroU32::new(3).expect("3 is non-zero")), + Egress::Direct, ), ( ExitId::new(NonZeroU8::new(2).expect("2 is non-zero")), ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")), + Egress::Direct, ), ];