Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497use std::future::Future;use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};use std::pin::Pin;use std::sync::Arc;use std::time::Duration;
use bytes::Bytes;use futures::TryStreamExt;use http::{HeaderMap, Method, StatusCode};use url::{Host, Url};
#[derive(Debug, Clone, thiserror::Error)]pub enum NetworkError { #[error("build: {0}")] Build(String), #[error("connect: {0}")] Connect(String), #[error("timeout: {0}")] Timeout(String), #[error("request: {0}")] Request(String), #[error("body: {0}")] Body(String), #[error("response exceeds {limit} bytes")] TooLarge { limit: u64 }, #[error("refusing to reach non-public address {host}")] Blocked { host: String },}
pub fn is_blocked_ip(ip: IpAddr) -> bool { match ip { IpAddr::V4(v4) => is_blocked_v4(v4), IpAddr::V6(v6) => match embedded_ipv4(v6) { Some(embedded) => is_blocked_v4(embedded), None => { v6.is_loopback() || v6.is_unspecified() || v6.is_multicast() || (v6.segments()[0] & 0xfe00) == 0xfc00 || (v6.segments()[0] & 0xffc0) == 0xfe80 } }, }}
fn is_blocked_v4(v4: Ipv4Addr) -> bool { v4.is_loopback() || v4.is_private() || v4.is_link_local() || v4.is_unspecified() || v4.is_broadcast() || v4.is_documentation() || v4.is_multicast() || v4.octets()[0] == 0 || v4.octets()[0] >= 240 || matches!(v4.octets(), [100, second, ..] if (64..=127).contains(&second)) || matches!(v4.octets(), [198, second, ..] if (18..=19).contains(&second))}
fn embedded_ipv4(v6: Ipv6Addr) -> Option<Ipv4Addr> { if let Some(mapped) = v6.to_ipv4() { return Some(mapped); } let segments = v6.segments(); if segments[0] == 0x2002 { return Some(Ipv4Addr::new( (segments[1] >> 8) as u8, segments[1] as u8, (segments[2] >> 8) as u8, segments[2] as u8, )); } if segments[0] == 0x0064 && segments[1] == 0xff9b && segments[2..6] == [0, 0, 0, 0] { return Some(Ipv4Addr::new( (segments[6] >> 8) as u8, segments[6] as u8, (segments[7] >> 8) as u8, segments[7] as u8, )); } None}
struct GuardedResolver;
impl reqwest::dns::Resolve for GuardedResolver { fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving { Box::pin(async move { let host = name.as_str().to_owned(); let resolved = tokio::net::lookup_host((host.as_str(), 0)).await?; let allowed: Vec<SocketAddr> = resolved.filter(|addr| !is_blocked_ip(addr.ip())).collect(); if allowed.is_empty() { return Err(Box::<dyn std::error::Error + Send + Sync>::from(format!( "{host} resolves only to non-public addresses" ))); } Ok(Box::new(allowed.into_iter()) as reqwest::dns::Addrs) }) }}
#[derive(Debug, Clone, Copy)]pub struct HttpLimits { pub connect_timeout: Duration, pub read_timeout: Duration, pub request_timeout: Duration, pub max_response_bytes: u64, pub block_private_addresses: bool,}
impl Default for HttpLimits { fn default() -> Self { Self { connect_timeout: Duration::from_secs(5), read_timeout: Duration::from_secs(30), request_timeout: Duration::from_secs(60), max_response_bytes: 16 * 1024 * 1024, block_private_addresses: true, } }}
pub struct HttpRequest { pub method: Method, pub url: Url, pub headers: HeaderMap, pub body: Option<Bytes>,}
impl HttpRequest { pub fn get(url: Url) -> Self { Self { method: Method::GET, url, headers: HeaderMap::new(), body: None, } }
pub fn post(url: Url, body: Bytes) -> Self { Self { method: Method::POST, url, headers: HeaderMap::new(), body: Some(body), } }}
#[derive(Debug)]pub struct HttpResponse { pub status: StatusCode, pub headers: HeaderMap, pub body: Bytes,}
pub type HttpFuture = Pin<Box<dyn Future<Output = Result<HttpResponse, NetworkError>> + Send>>;
pub type ByteStream = Pin<Box<dyn futures::Stream<Item = Result<Bytes, NetworkError>> + Send>>;
pub struct StreamedResponse { pub status: StatusCode, pub headers: HeaderMap, pub body: ByteStream,}
pub type StreamFuture = Pin<Box<dyn Future<Output = Result<StreamedResponse, NetworkError>> + Send>>;
pub trait HttpTransport: Send + Sync + 'static { fn execute(&self, request: HttpRequest) -> HttpFuture;
fn execute_streamed(&self, request: HttpRequest) -> StreamFuture { let response = self.execute(request); Box::pin(async move { let response = response.await?; Ok(StreamedResponse { status: response.status, headers: response.headers, body: Box::pin(futures::stream::once(std::future::ready(Ok(response.body)))), }) }) }}
pub struct ReqwestHttp { client: reqwest::Client, max_response_bytes: u64, block_private_addresses: bool,}
impl ReqwestHttp { pub fn new(limits: HttpLimits) -> Result<Self, NetworkError> { let mut builder = reqwest::Client::builder() .connect_timeout(limits.connect_timeout) .read_timeout(limits.read_timeout) .timeout(limits.request_timeout) .redirect(reqwest::redirect::Policy::none()); if limits.block_private_addresses { builder = builder.dns_resolver(Arc::new(GuardedResolver)); } if let Some(path) = std::env::var_os("KNOT_EXTRA_CA_FILE") { let pem = std::fs::read(&path).map_err(|error| NetworkError::Build(error.to_string()))?; builder = reqwest::Certificate::from_pem_bundle(&pem) .map_err(|error| NetworkError::Build(error.to_string()))? .into_iter() .fold(builder, reqwest::ClientBuilder::add_root_certificate); } let client = builder .build() .map_err(|error| NetworkError::Build(error.to_string()))?; Ok(Self { client, max_response_bytes: limits.max_response_bytes, block_private_addresses: limits.block_private_addresses, }) }}
impl HttpTransport for ReqwestHttp { fn execute(&self, request: HttpRequest) -> HttpFuture { let client = self.client.clone(); let limit = self.max_response_bytes; let guard = self.block_private_addresses; Box::pin(async move { if let Some(host) = guard.then(|| blocked_literal(&request.url)).flatten() { return Err(NetworkError::Blocked { host }); } let mut builder = client .request(request.method, request.url) .headers(request.headers); if let Some(body) = request.body { builder = builder.body(body); } let response = builder.send().await.map_err(map_reqwest)?; let status = response.status(); let headers = response.headers().clone(); if response.content_length().is_some_and(|len| len > limit) { return Err(NetworkError::TooLarge { limit }); } let body = bounded_body(response, limit).await?; Ok(HttpResponse { status, headers, body, }) }) }
fn execute_streamed(&self, request: HttpRequest) -> StreamFuture { let client = self.client.clone(); let guard = self.block_private_addresses; Box::pin(async move { if let Some(host) = guard.then(|| blocked_literal(&request.url)).flatten() { return Err(NetworkError::Blocked { host }); } let mut builder = client .request(request.method, request.url) .headers(request.headers); if let Some(body) = request.body { builder = builder.body(body); } let response = builder.send().await.map_err(map_reqwest)?; let status = response.status(); let headers = response.headers().clone(); let body: ByteStream = Box::pin(response.bytes_stream().map_err(|error| { if error.is_timeout() { NetworkError::Timeout(error.to_string()) } else { NetworkError::Body(error.to_string()) } })); Ok(StreamedResponse { status, headers, body, }) }) }}
async fn bounded_body(response: reqwest::Response, limit: u64) -> Result<Bytes, NetworkError> { response .bytes_stream() .map_err(|error| { if error.is_timeout() { NetworkError::Timeout(error.to_string()) } else { NetworkError::Body(error.to_string()) } }) .try_fold(Vec::new(), |mut buffer, chunk| async move { if buffer.len() as u64 + chunk.len() as u64 > limit { return Err(NetworkError::TooLarge { limit }); } buffer.extend_from_slice(&chunk); Ok(buffer) }) .await .map(Bytes::from)}
fn blocked_literal(url: &Url) -> Option<String> { match url.host()? { Host::Ipv4(ip) if is_blocked_ip(IpAddr::V4(ip)) => Some(ip.to_string()), Host::Ipv6(ip) if is_blocked_ip(IpAddr::V6(ip)) => Some(ip.to_string()), _ => None, }}
fn map_reqwest(error: reqwest::Error) -> NetworkError { if error.is_timeout() { NetworkError::Timeout(error.to_string()) } else if error.is_connect() { NetworkError::Connect(error.to_string()) } else { NetworkError::Request(error.to_string()) }}
pub struct FakeHttp<F> { responder: F,}
impl<F> FakeHttp<F>where F: Fn(&HttpRequest) -> Result<HttpResponse, NetworkError> + Send + Sync + 'static,{ pub fn new(responder: F) -> Self { Self { responder } }}
impl<F> HttpTransport for FakeHttp<F>where F: Fn(&HttpRequest) -> Result<HttpResponse, NetworkError> + Send + Sync + 'static,{ fn execute(&self, request: HttpRequest) -> HttpFuture { let result = (self.responder)(&request); Box::pin(async move { result }) }}
#[cfg(test)]mod tests { use super::*; use futures::StreamExt; use std::io::{Read, Write}; use std::net::{SocketAddr, TcpListener};
#[test] fn blocked_addresses_cover_the_internal_ranges() { let blocked = [ "127.0.0.1", "10.0.0.5", "192.168.1.1", "172.16.0.1", "169.254.169.254", "100.64.0.1", "198.18.0.1", "240.0.0.1", "0.0.0.0", "::1", "::ffff:127.0.0.1", "fd00::1", "fe80::1", "2002:7f00:1::", "64:ff9b::7f00:1", ]; for raw in blocked { assert!( is_blocked_ip(raw.parse().unwrap()), "{raw} should be blocked" ); } let allowed = [ "1.1.1.1", "8.8.8.8", "93.184.216.34", "2606:4700:4700::1111", "2002:808:808::", "64:ff9b::808:808", ]; for raw in allowed { assert!( !is_blocked_ip(raw.parse().unwrap()), "{raw} should be allowed" ); } }
fn tiny_limits(max_response_bytes: u64, request_timeout: Duration) -> HttpLimits { HttpLimits { connect_timeout: Duration::from_millis(200), read_timeout: Duration::from_millis(200), request_timeout, max_response_bytes, block_private_addresses: false, } }
fn serve_body(body: Vec<u8>, with_content_length: bool) -> SocketAddr { let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); let addr = listener.local_addr().expect("addr"); std::thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { let _ = stream.read(&mut [0u8; 1024]); let header = if with_content_length { format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()) } else { "HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n".to_string() }; let _ = stream.write_all(header.as_bytes()); let _ = stream.write_all(&body); } }); addr }
fn serve_hang() -> SocketAddr { let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); let addr = listener.local_addr().expect("addr"); std::thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { let _ = stream.read(&mut [0u8; 1024]); std::thread::sleep(Duration::from_secs(5)); drop(stream); } }); addr }
async fn fetch(addr: SocketAddr, limits: HttpLimits) -> Result<HttpResponse, NetworkError> { let transport = ReqwestHttp::new(limits).expect("client builds"); let url = Url::parse(&format!("http://{addr}/")).expect("url"); transport.execute(HttpRequest::get(url)).await }
#[tokio::test] async fn oversized_response_is_rejected_whether_declared_or_streamed() { futures::stream::iter([true, false]) .for_each(|with_content_length| async move { let addr = serve_body(vec![0u8; 4096], with_content_length); let result = fetch(addr, tiny_limits(64, Duration::from_secs(2))).await; assert!(matches!(result, Err(NetworkError::TooLarge { limit: 64 }))); }) .await; }
#[tokio::test] async fn small_response_within_limit_succeeds() { let addr = serve_body(b"pong".to_vec(), true); let response = fetch(addr, tiny_limits(64, Duration::from_secs(2))) .await .expect("response within limit"); assert_eq!(response.body.as_ref(), b"pong"); }
#[tokio::test] async fn unresponsive_server_times_out() { let addr = serve_hang(); let result = fetch(addr, tiny_limits(1024, Duration::from_millis(150))).await; assert!(matches!(result, Err(NetworkError::Timeout(_)))); }
#[tokio::test] async fn connect_failure_surfaces_typed_error() { let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); let addr = listener.local_addr().expect("addr"); drop(listener); let result = fetch(addr, tiny_limits(1024, Duration::from_secs(2))).await; assert!(matches!( result, Err(NetworkError::Connect(_) | NetworkError::Request(_) | NetworkError::Timeout(_)) )); }
#[test] fn fake_http_returns_canned_response() { let transport = FakeHttp::new(|request: &HttpRequest| { assert_eq!(request.method, Method::GET); Ok(HttpResponse { status: StatusCode::OK, headers: HeaderMap::new(), body: Bytes::from_static(b"pong"), }) }); let request = HttpRequest::get(Url::parse("https://oyster.cafe/ping").unwrap()); let response = futures::executor::block_on(transport.execute(request)).expect("fake response"); assert_eq!(response.status, StatusCode::OK); assert_eq!(response.body.as_ref(), b"pong"); }}