[ssrf] couple public egress validation to sockets
1535 unmodified linesExpand all1536153715381539154015411542154315443297 unmodified linesExpand all1535 unmodified linesExpand all "hex", "http", "humantime", "hyper", "ipnet", "iroh-car", "jacquard-api", "jacquard-common", "jacquard-derive",3297 unmodified linesExpand all38 unmodified lines39404142434445464759 unmodified lines38 unmodified lineszstd = { version = "0.13", optional = true }
bytes = { version = "1.11.1", features = ["serde"] }url = { version = "2.5", features = ["serde"] }ipnet = "2.12.0"smol_str = "0.3"futures = "0.3"reqwest = { version = "0.12.8", features = ["json", "rustls-tls-native-roots", "stream", "gzip", "brotli", "zstd", "http2"], default-features = false }axum = { version = "0.8.8", features = ["macros"] }59 unmodified lines123456789101112131415161718192021The MIT License
Copyright (c) 2021 Svix (https://www.svix.com)
Permission is hereby granted, free of charge, to any person obtaining a copyof this software and associated documentation files (the "Software"), to dealin the Software without restriction, including without limitation the rightsto use, copy, modify, merge, publish, distribute, sublicense, and/or sellcopies of the Software, and to permit persons to whom the Software isfurnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included inall copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS ORIMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THEAUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHERLIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS INTHE SOFTWARE.448 unmodified linesExpand all449450451452453454455456457458459460461462463262 unmodified linesExpand all448 unmodified linesExpand all "the backlinks feature is not supported with HYDRANT_EPHEMERAL / ephemeral" )); }
#[cfg(feature = "indexer")] if !self.backfill_proxies.is_empty() { return Err(miette::miette!( "HYDRANT_BACKFILL_PROXIES is unsupported: public backfill egress requires direct sockets" )); }
#[cfg(feature = "relay")] if self.only_index_links { return Err(miette::miette!( "HYDRANT_ONLY_INDEX_LINKS is not supported in relay mode"262 unmodified linesExpand all43 unmodified lines4445464748495051527 unmodified lines43 unmodified linespub(crate) mod db;pub(crate) mod ingest;#[cfg(feature = "jetstream")]pub(crate) mod jetstream;pub(crate) mod net;#[cfg(feature = "indexer")]pub(crate) mod ops;pub(crate) mod patch;pub(crate) mod pds_discovery;7 unmodified lines1123434566789101112131415161718192021222324252627282930313233343536373839404142434445464774950515253545556575859606162636465666768697071727374757677787980818283848586878889909192938910961112131415161718193 unmodified lines23242526108272829303132333435113114115116117363738395 unmodified lines4546474813113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817949505152531851861871881891901911921931941951961971981992002012022032042052062072082092102112122135455565710 unmodified linesuse std::net::IpAddr;use url::{Host, Url};
use thiserror::Error;use url::Url;#[cfg(feature = "indexer")]pub(crate) use crate::net::is_private_ip;pub(crate) use crate::net::{HostError as PdsHostError, PublicHost as CanonicalPdsHost};
/// a normalized public PDS host suitable for a direct firehose URL.#[derive(Debug, Clone, PartialEq, Eq, Hash)]pub(crate) struct CanonicalPdsHost(String);
impl CanonicalPdsHost { pub(crate) fn as_str(&self) -> &str { &self.0 }
pub(crate) fn firehose_url(&self) -> Url { let mut url = Url::parse("wss://localhost/").expect("constant PDS URL must parse"); let host = self .0 .parse::<std::net::Ipv6Addr>() .map(|_| format!("[{}]", self.0)) .unwrap_or_else(|_| self.0.clone()); url.set_host(Some(&host)) .expect("canonical PDS host must be valid in a URL"); url }}
#[derive(Debug, Clone, PartialEq, Eq, Error)]pub(crate) enum PdsHostError { #[error("PDS hostname is empty")] Empty, #[error("PDS hostname must not include a port, path, query, fragment, or userinfo")] NotHostname, #[error("PDS hostname is localhost or a local-only domain")] LocalName, #[error("PDS hostname is a non-public IP address")] NonPublicIp, #[error("PDS hostname is not a valid DNS name")] InvalidDnsName, #[error("PDS source URL must use wss with no port, path, query, fragment, or userinfo")] NonCanonicalUrl,}
/// normalize an untrusted PDS hostname without resolving it.////// this rejects syntactically unsafe names and literal non-public addresses. it/// deliberately does not resolve DNS, so it cannot prevent DNS rebinding.pub(crate) fn canonical_pds_host(input: &str) -> Result<CanonicalPdsHost, PdsHostError> { let host = input.strip_suffix('.').unwrap_or(input); if host.is_empty() { return Err(PdsHostError::Empty); } if host.contains('@') || (host.starts_with('[') && !host.ends_with(']')) { return Err(PdsHostError::NotHostname); }
let literal_ip = host .strip_prefix('[') .and_then(|host| host.strip_suffix(']')) .unwrap_or(host); if let Ok(ip) = literal_ip.parse::<IpAddr>() { return (!is_private_ip(ip)) .then(|| CanonicalPdsHost(ip.to_string())) .ok_or(PdsHostError::NonPublicIp); } if host.contains(':') { return Err(PdsHostError::NotHostname); }
let parsed = Url::parse(&format!("wss://{host}/")).map_err(|_| PdsHostError::InvalidDnsName)?; if !parsed.username().is_empty() || parsed.password().is_some() || parsed.port().is_some() || parsed.path() != "/" || parsed.query().is_some() || parsed.fragment().is_some() { return Err(PdsHostError::NotHostname); } let host = parsed.host_str().ok_or(PdsHostError::InvalidDnsName)?; if let Ok(ip) = host.parse::<IpAddr>() { return (!is_private_ip(ip)) .then(|| CanonicalPdsHost(ip.to_string())) .ok_or(PdsHostError::NonPublicIp); } if host == "localhost" || host.ends_with(".local") || host.ends_with(".internal") { return Err(PdsHostError::LocalName); } if !is_dns_name(host) { return Err(PdsHostError::InvalidDnsName); }
Ok(CanonicalPdsHost(host.to_string())) crate::net::PublicHost::parse(input)}
/// validate and normalize a direct-PDS source URL./// validate and normalize the canonical direct-PDS websocket source URL.////// This remains the storage/admission canonicalization law. Public HTTP/// endpoints use [`crate::net::PublicEndpoint`] instead because XRPC paths and/// non-default ports are meaningful there.pub(crate) fn canonical_pds_url(url: &Url) -> Result<(CanonicalPdsHost, Url), PdsHostError> { if url.scheme() != "wss" || !url.username().is_empty() || url.password().is_some()3 unmodified lines || url.fragment().is_some() { return Err(PdsHostError::NonCanonicalUrl); } let host = canonical_pds_host(url.host_str().ok_or(PdsHostError::NonCanonicalUrl)?)?; let host = match url.host().ok_or(PdsHostError::NonCanonicalUrl)? { Host::Domain(host) => canonical_pds_host(host)?, Host::Ipv4(ip) => canonical_pds_host(&ip.to_string())?, Host::Ipv6(ip) => canonical_pds_host(&ip.to_string())?, }; let normalized = host.firehose_url(); Ok((host, normalized))}
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]pub(crate) fn is_safe_seed_host(host: &str) -> bool { canonical_pds_host(host).is_ok()}
pub(crate) fn list_hosts_url(base: &Url, cursor: Option<&str>) -> Url { let mut url = base.clone(); url.set_path("/xrpc/com.atproto.sync.listHosts"); {5 unmodified lines } url}
pub(crate) fn is_private_ip(ip: IpAddr) -> bool { match ip { IpAddr::V4(ip) => { let [a, b, c, _] = ip.octets(); ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_multicast() || ip.is_unspecified() || ip.is_broadcast() || a == 0 || (a == 100 && b & 0xc0 == 0x40) || (a == 192 && b == 0 && c == 0) || (a == 192 && b == 0 && c == 2) || (a == 192 && b == 88 && c == 99) || (a == 198 && b & 0xfe == 18) || (a == 198 && b == 51 && c == 100) || (a == 203 && b == 0 && c == 113) || a >= 240 } IpAddr::V6(ip) => { let octets = ip.octets(); ip.is_loopback() || ip.is_multicast() || ip.is_unspecified() || ip .to_ipv4_mapped() .is_some_and(|ipv4| is_private_ip(IpAddr::V4(ipv4))) || (octets[0] & 0xfe) == 0xfc || (octets[0] == 0xfe && octets[1] & 0xc0 == 0x80) || (octets[0] == 0xfe && octets[1] & 0xc0 == 0xc0) || (octets[..4] == [0x20, 0x01, 0x0d, 0xb8]) } }}
fn is_dns_name(host: &str) -> bool { host.len() <= 253 && host.split('.').all(|label| { !label.is_empty() && label.len() <= 63 && !label.starts_with('-') && !label.ends_with('-') && label .bytes() .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') })}
#[cfg(test)]mod tests { use super::*;
#[test] fn rejects_non_public_address_ranges() { for address in [ "0.0.0.1", "10.0.0.1", "100.64.0.1", "127.0.0.1", "169.254.1.1", "192.0.2.1", "198.18.0.1", "198.51.100.1", "203.0.113.1", "240.0.0.1", "::1", "::ffff:127.0.0.1", "2001:db8::1", "fc00::1", "fe80::1", "fec0::1", ] { assert!( is_private_ip(address.parse().unwrap()), "accepted {address}" ); } assert!(!is_private_ip("1.1.1.1".parse().unwrap())); assert!(!is_private_ip("2606:4700:4700::1111".parse().unwrap())); }
#[test] fn canonicalizes_public_dns_hosts_and_rejects_private_hosts() { let host = canonical_pds_host("BÜCHER.Example.").unwrap(); assert_eq!(host.as_str(), "xn--bcher-kva.example"); assert_eq!(host.firehose_url().as_str(), "wss://xn--bcher-kva.example/");10 unmodified lines14 unmodified lines15161718191920212223242526272845 unmodified lines7475767778787980818217 unmodified lines1001011021031041041051061071081091101071111121131141153 unmodified lines11912012112212312412512612712812913013113213313413513612412512612712813713813113914014114214317 unmodified lines161162163164157165166167168169167 unmodified linesExpand all3373383393403333413423433443456 unmodified lines3523533543553483563573583513593603613623631 unmodified line3653663673683613693703713723733743753763773783793803813823833843853863873883893903913923933949 unmodified lines40440540640738138240840941041141241341441541641728 unmodified lines4464474484494194504514524534544554564574584594604614624274284634644654664674684694705 unmodified lines14 unmodified lines IdentityError, IdentityErrorKind, IdentityResolver, PlcSource, ResolverOptions,};use miette::{Diagnostic, IntoDiagnostic};use reqwest::StatusCode;use reqwest::redirect::Policy;use scc::HashCache;use smol_str::SmolStr;use thiserror::Error;use url::Url;
use crate::net::{PublicHttpClient, public_http};use crate::util::url_to_fluent_uri;
// as per spec: https://web.plc.directory/spec/v0.1/did-plc// "As an anti-abuse mechanism, operations have a maximum size when encoded as45 unmodified lines pub key: Option<PublicKey<'static>>,}
struct ResolverInner { jacquards: Vec<JacquardResolver<reqwest::Client>>, jacquards: Vec<JacquardResolver<PublicHttpClient>>, next_idx: AtomicUsize, cache: HashCache<Did, MiniDoc>,}
17 unmodified lines Self::with_options(options, identity_cache_size) }
#[cfg(all(test, feature = "indexer_stream"))] pub(crate) fn slingshot_for_test(base: Url, identity_cache_size: u64) -> Self { pub(crate) fn slingshot_for_test( base: Url, identity_cache_size: u64, public: PublicHttpClient, ) -> Self { use jacquard_identity::resolver::{DidStep, HandleStep};
Self::with_options( Self::with_options_and_client( vec![ResolverOptions { plc_source: PlcSource::Slingshot { base: url_to_fluent_uri(&base), },3 unmodified lines request_timeout: Some(Duration::from_secs(3)), ..Default::default() }], identity_cache_size, public, ) }
fn with_options(options: Vec<ResolverOptions>, identity_cache_size: u64) -> Self { Self::with_options_and_client(options, identity_cache_size, public_http()) }
pub(crate) fn with_options_and_client( options: Vec<ResolverOptions>, identity_cache_size: u64, public: PublicHttpClient, ) -> Self { assert!(!options.is_empty(), "at least one PLC URL must be provided"); let http = reqwest::Client::builder() .redirect(Policy::none()) .timeout(Duration::from_secs(20)) .build() .expect("resolver HTTP client configuration should be valid"); let jacquards = options .into_iter() .map(|options| JacquardResolver::new(http.clone(), options).with_system_dns()) .map(|options| JacquardResolver::new(public.clone(), options).with_system_dns()) .collect();
Self { inner: Arc::new(ResolverInner {17 unmodified lines
async fn req<'r, T, Fut>( &'r self, is_plc: bool, f: impl Fn(&'r JacquardResolver<reqwest::Client>) -> Fut, f: impl Fn(&'r JacquardResolver<PublicHttpClient>) -> Fut, ) -> Result<T, ResolverError> where Fut: Future<Output = Result<T, IdentityError>>, {167 unmodified linesExpand all ResolverError::Ratelimited )); }
async fn spawn_mutable_doc_server(body: String) -> (Url, Arc<RwLock<String>>) { async fn spawn_mutable_doc_server(body: String) -> (std::net::SocketAddr, Arc<RwLock<String>>) { let current = Arc::new(RwLock::new(body)); let served = current.clone(); let app = Router::new().fallback(move || { let served = served.clone();6 unmodified lines let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); (Url::parse(&format!("http://{addr}/")).unwrap(), current) (addr, current) }
async fn spawn_doc_server(body: String) -> Url { async fn spawn_doc_server(body: String) -> std::net::SocketAddr { let app = Router::new().fallback(move || { let body = body.clone(); async move { ([(header::CONTENT_TYPE, "application/json")], body).into_response() } });1 unmodified line let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); Url::parse(&format!("http://{addr}/")).unwrap() addr }
fn test_public_client(address: std::net::SocketAddr) -> PublicHttpClient { let client = reqwest::Client::builder() .no_proxy() .resolve("public.example", address) .build() .unwrap(); PublicHttpClient::test(client) }
fn test_options() -> ResolverOptions { ResolverOptions { plc_source: PlcSource::PlcDirectory { base: url_to_fluent_uri(&Url::parse("http://public.example/").unwrap()), }, request_timeout: Some(Duration::from_secs(3)), ..Default::default() } }
#[tokio::test] async fn fresh_resolution_bypasses_cache_until_explicitly_installed() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let doc = |pds: &str| {9 unmodified lines "verificationMethod": [] }) .to_string() }; let (url, current) = spawn_mutable_doc_server(doc("https://pds-a.example")).await; let resolver = Resolver::new(vec![url], 10); let (address, current) = spawn_mutable_doc_server(doc("https://pds-a.example")).await; let resolver = Resolver::with_options_and_client( vec![test_options()], 10, test_public_client(address), );
assert_eq!( resolver.resolve_doc(&did).await.unwrap().pds.as_str(), "https://pds-a.example/"28 unmodified lines }], "verificationMethod": [] }) .to_string(); let resolver = Resolver::new(vec![spawn_doc_server(body).await], 10); let address = spawn_doc_server(body).await; let resolver = Resolver::with_options_and_client( vec![test_options()], 10, test_public_client(address), );
let (data, handle) = resolver.resolve_raw_doc(&did).await.unwrap(); let json = serde_json::to_value(data).unwrap(); assert_eq!(json["id"], did.as_str()); assert_eq!(json["service"][0]["serviceEndpoint"], "https://pds.example"); assert_eq!(handle.unwrap().as_str(), "alice.test");
let oversized = Resolver::new( vec![spawn_doc_server("x".repeat(MAX_DID_DOC_BYTES + 1)).await], let oversized = Resolver::with_options_and_client( vec![test_options()], 10, test_public_client(spawn_doc_server("x".repeat(MAX_DID_DOC_BYTES + 1)).await), ); let error = oversized.resolve_raw_doc(&did).await.unwrap_err(); assert!( error5 unmodified lines165 unmodified linesExpand all166167168169170171172173170171172173174175176257 unmodified linesExpand all165 unmodified linesExpand all verify_cids: config.verify_cids, max_car_body_bytes: config.max_car_body_bytes, get_repo_semaphore: Semaphore::new(config.get_repo_concurrency_limit), maintenance_timeout: config.maintenance_timeout, throttler: Throttler::new( config.backfill_proxies.len() + 1, config.per_pds_concurrency, ), // public backfill is direct-only; one socket pool must not inflate // per-PDS pacing as if proxies provided independent egress IPs. throttler: Throttler::new(1, config.per_pds_concurrency), pds_daily_limit, }) }
257 unmodified linesExpand all