//! socket-coupled egress capabilities for externally influenced endpoints.
//! [`PublicEndpoint`] carries syntax proof only. The public HTTP and websocket
//! resolvers carry the address proof to the connector: every fresh connection
//! resolves the complete answer set, rejects it if any answer is not public, and
//! only then gives the addresses to the socket implementation.
//! the IPv4/IPv6 classifier below is adapted from Svix's `is_allowed` and helper
//! logic, pinned to commit `805721db7165a00275c92460fae4363f2fb9b720` in
//! `svix-webhooks/server/svix-server/src/core/webhook_http_client.rs`. Svix is
//! MIT licensed; its exact notice is retained in
//! `licenses/svix-webhooks-MIT.txt`. The range policy and its mapped-IPv6
//! canonicalization are intentionally retained rather than replaced with an
//! ad-hoc decoder.
use std::future::Future;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use futures::StreamExt;
use ipnet::{Ipv4Net, Ipv6Net};
use reqwest::dns::{Addrs, Name, Resolve, Resolving};
use reqwest::redirect::Policy;
use smol_str::SmolStr;
use thiserror::Error;
use url::{Host, Url};
pub(crate) const DNS_LOOKUP_TIMEOUT: Duration = Duration::from_secs(5);
const PUBLIC_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// a peer that drops our SYN answers nothing, so a blackholed connect would
/// otherwise spend the whole request budget before the next candidate source
/// gets a turn. applies to every public client, and never raised above the
/// caller's own request timeout.
const PUBLIC_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
/// handle resolution runs on the read path of every mini doc, and one of its
/// steps fetches a host the account itself names, so it gets its own budget.
pub(crate) const HANDLE_REQUEST_TIMEOUT: Duration = Duration::from_secs(3);
pub(crate) const HANDLE_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
/// did docs gate ingest, so a host that hangs has to fail fast instead of holding a lookup for
/// the whole public budget
pub(crate) const IDENTITY_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
const IDENTITY_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
/// A hostname or public IP literal accepted by the public egress syntax policy.
///
/// This value contains no port, path, query, fragment, or userinfo. DNS names
/// are lower-case IDNA ASCII and have no trailing dot. A literal is accepted
/// only when its address is public according to [`is_private_ip`].
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) struct PublicHost(SmolStr);
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub(crate) enum HostError {
#[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,
}
impl PublicHost {
/// Parse a host authority without resolving it.
pub(crate) fn parse(input: &str) -> Result<Self, HostError> {
let host = input.strip_suffix('.').unwrap_or(input);
if host.is_empty() {
return Err(HostError::Empty);
}
if host.contains('@') || (host.starts_with('[') && !host.ends_with(']')) {
return Err(HostError::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(|| Self(ip.to_string().into()))
.ok_or(HostError::NonPublicIp);
}
if host.contains(':') {
return Err(HostError::NotHostname);
}
// Url owns IDNA and authority parsing. Constructing a host-only URL
// also rejects malformed percent escapes and authority syntax.
let parsed =
Url::parse(&format!("wss://{host}/")).map_err(|_| HostError::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(HostError::NotHostname);
}
let host = parsed.host_str().ok_or(HostError::InvalidDnsName)?;
if let Ok(ip) = host.parse::<IpAddr>() {
return (!is_private_ip(ip))
.then(|| Self(ip.to_string().into()))
.ok_or(HostError::NonPublicIp);
}
if matches!(host, "localhost" | "local" | "internal")
|| host.ends_with(".localhost")
|| host.ends_with(".local")
|| host.ends_with(".internal")
{
return Err(HostError::LocalName);
}
if !is_dns_name(host) {
return Err(HostError::InvalidDnsName);
}
Ok(Self(host.into()))
}
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::<Ipv6Addr>()
.map(|_| format!("[{}]", self.0))
.unwrap_or_else(|_| self.0.to_string());
url.set_host(Some(&host))
.expect("public host must be valid in a URL");
url
}
/// Construct the canonical HTTPS base URL from this already-validated host.
/// URL serialization owns IPv6 bracket handling; callers do not format an
/// authority string by hand.
#[cfg(feature = "indexer")]
pub(crate) fn https_endpoint(&self) -> PublicEndpoint {
let mut url = Url::parse("https://localhost/").expect("constant PDS URL must parse");
let host = self
.0
.parse::<Ipv6Addr>()
.map(|_| format!("[{}]", self.0))
.unwrap_or_else(|_| self.0.to_string());
url.set_host(Some(&host))
.expect("public host must be valid in a URL");
PublicEndpoint { url }
}
}
/// A URL whose authority passed the public syntax policy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PublicEndpoint {
url: Url,
}
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub(crate) enum EndpointError {
#[error("public endpoint scheme is not supported: {0}")]
UnsupportedScheme(SmolStr),
#[error("public endpoint must not include userinfo or a fragment")]
UnsafeAuthority,
#[error("public endpoint has no host")]
MissingHost,
#[error(transparent)]
Host(#[from] HostError),
}
impl PublicEndpoint {
/// Parse a direct HTTP endpoint where a caller's protocol explicitly
/// permits cleartext HTTP. HTTPS remains accepted.
pub(crate) fn parse_http(url: &Url) -> Result<Self, EndpointError> {
Self::parse_schemes(url, |scheme| matches!(scheme, "http" | "https"))
}
/// Parse the public websocket endpoint. Public firehose connections are
/// TLS-only; the trusted raw path remains available to operator config.
pub(crate) fn parse_websocket(url: &Url) -> Result<Self, EndpointError> {
Self::parse_schemes(url, |scheme| scheme == "wss")
}
fn parse_schemes(
url: &Url,
supported: impl FnOnce(&str) -> bool,
) -> Result<Self, EndpointError> {
if !supported(url.scheme()) {
return Err(EndpointError::UnsupportedScheme(url.scheme().into()));
}
if !url.username().is_empty() || url.password().is_some() || url.fragment().is_some() {
return Err(EndpointError::UnsafeAuthority);
}
// use Url::host rather than host_str so bracketed IPv6 and mapped IPv6
// literals are represented as addresses before the shared classifier runs.
let host = match url.host().ok_or(EndpointError::MissingHost)? {
Host::Domain(host) => PublicHost::parse(host)?,
Host::Ipv4(ip) => PublicHost::parse(&ip.to_string())?,
Host::Ipv6(ip) => PublicHost::parse(&ip.to_string())?,
};
let mut normalized = url.clone();
let host_for_url = host
.as_str()
.parse::<Ipv6Addr>()
.map(|_| format!("[{}]", host.as_str()))
.unwrap_or_else(|_| host.as_str().to_owned());
normalized
.set_host(Some(&host_for_url))
.map_err(|_| EndpointError::Host(HostError::InvalidDnsName))?;
Ok(Self { url: normalized })
}
pub(crate) fn url(&self) -> &Url {
&self.url
}
#[cfg(feature = "indexer")]
pub(crate) fn with_path(&self, path: &str) -> Self {
let mut endpoint = self.clone();
endpoint.url.set_path(path);
endpoint
}
#[cfg(feature = "indexer")]
pub(crate) fn append_query_pair(&mut self, key: &str, value: &str) {
self.url.query_pairs_mut().append_pair(key, value);
}
}
/// The system resolver used by both public HTTP and websocket connectors.
///
/// Every lookup validates the complete answer set before returning any address.
/// The injected lookup makes the policy independently testable without live DNS.
#[derive(Clone)]
pub(crate) struct PublicResolver {
lookup: Arc<dyn Lookup>,
}
trait Lookup: Send + Sync {
fn lookup(
&self,
host: String,
port: u16,
) -> Pin<Box<dyn Future<Output = Result<Vec<SocketAddr>, String>> + Send>>;
}
struct SystemLookup;
impl Lookup for SystemLookup {
fn lookup(
&self,
host: String,
port: u16,
) -> Pin<Box<dyn Future<Output = Result<Vec<SocketAddr>, String>> + Send>> {
Box::pin(async move {
tokio::net::lookup_host((host, port))
.await
.map(|addresses| addresses.collect())
.map_err(|error| error.to_string())
})
}
}
impl PublicResolver {
pub(crate) fn system() -> Self {
Self {
lookup: Arc::new(SystemLookup),
}
}
async fn resolve_addresses(
&self,
host: &str,
port: u16,
) -> Result<Vec<SocketAddr>, ResolveError> {
let addresses = tokio::time::timeout(
DNS_LOOKUP_TIMEOUT,
self.lookup.lookup(host.to_owned(), port),
)
.await
.map_err(|_| ResolveError::Timeout)?
.map_err(ResolveError::Lookup)?;
validate_addresses(addresses)
}
}
#[derive(Debug, Error)]
enum ResolveError {
#[error("DNS lookup timed out")]
Timeout,
#[error("DNS lookup failed: {0}")]
Lookup(String),
#[error("DNS lookup returned no addresses")]
Empty,
#[error("DNS answer contains non-public address: {0}")]
NonPublic(IpAddr),
}
fn validate_addresses(addresses: Vec<SocketAddr>) -> Result<Vec<SocketAddr>, ResolveError> {
if addresses.is_empty() {
return Err(ResolveError::Empty);
}
if let Some(address) = addresses.iter().find(|address| is_private_ip(address.ip())) {
return Err(ResolveError::NonPublic(address.ip()));
}
Ok(addresses)
}
impl Resolve for PublicResolver {
fn resolve(&self, name: Name) -> Resolving {
let resolver = self.clone();
let host = name.as_str().to_owned();
Box::pin(async move {
let addresses = resolver.resolve_addresses(&host, 0).await?;
Ok(Box::new(addresses.into_iter()) as Addrs)
})
}
}
impl tokio_websockets::resolver::Resolver for PublicResolver {
async fn resolve(&self, host: &str, port: u16) -> Result<SocketAddr, tokio_websockets::Error> {
self.resolve_addresses(host, port)
.await
.map_err(|_| tokio_websockets::Error::CannotResolveHost)?
.into_iter()
.next()
.ok_or(tokio_websockets::Error::CannotResolveHost)
}
}
const MAX_PUBLIC_HTTP_BODY_BYTES: usize = 2 * 1024 * 1024;
/// The only public HTTP request capability. Its raw reqwest client is private;
/// callers must provide a [`PublicEndpoint`] and cannot send an unchecked URL.
#[derive(Clone)]
pub(crate) struct PublicHttpClient {
client: Arc<reqwest::Client>,
}
#[derive(Debug, Error)]
pub enum PublicHttpError {
#[error(transparent)]
UnsafeEndpoint(#[from] EndpointError),
#[error(transparent)]
Reqwest(#[from] reqwest::Error),
#[error("public HTTP response body exceeds {limit} bytes")]
BodyTooLarge { limit: usize },
}
impl PublicHttpClient {
#[cfg(any(feature = "indexer", test))]
pub(crate) fn get(&self, endpoint: &PublicEndpoint) -> reqwest::RequestBuilder {
self.client.get(endpoint.url().clone())
}
/// Execute a streaming request after checking its final URL. The wrapped
/// client owns the socket-coupled resolver, so callers cannot substitute a
/// raw transport after obtaining endpoint syntax proof.
#[cfg(feature = "indexer")]
pub(crate) async fn execute(
&self,
request: reqwest::Request,
) -> Result<reqwest::Response, PublicHttpError> {
PublicEndpoint::parse_http(request.url()).map_err(PublicHttpError::UnsafeEndpoint)?;
self.client.execute(request).await.map_err(Into::into)
}
/// Send a Jacquard request only after proving its complete public endpoint
/// syntax. The response is consumed through a bounded decoded stream.
pub(crate) async fn send_http_bounded(
&self,
request: http::Request<Vec<u8>>,
max_body_bytes: usize,
) -> Result<http::Response<Vec<u8>>, PublicHttpError> {
send_public_http(&self.client, request, max_body_bytes).await
}
pub(crate) async fn send_http(
&self,
request: http::Request<Vec<u8>>,
) -> Result<http::Response<Vec<u8>>, PublicHttpError> {
self.send_http_bounded(request, MAX_PUBLIC_HTTP_BODY_BYTES)
.await
}
#[cfg(test)]
pub(crate) fn test(client: reqwest::Client) -> Self {
Self {
client: Arc::new(client),
}
}
}
/// Public transport for legacy XRPC adapters. Unlike reqwest's DNS hook alone,
/// this validates the request URI first because literal IPs bypass DNS. The
/// response is consumed through a bounded decoded stream.
async fn send_public_http(
client: &reqwest::Client,
request: http::Request<Vec<u8>>,
max_body_bytes: usize,
) -> Result<http::Response<Vec<u8>>, PublicHttpError> {
let url = Url::parse(&request.uri().to_string()).map_err(|_| {
PublicHttpError::UnsafeEndpoint(EndpointError::Host(HostError::InvalidDnsName))
})?;
PublicEndpoint::parse_http(&url).map_err(PublicHttpError::UnsafeEndpoint)?;
send_reqwest(client, request, max_body_bytes).await
}
async fn send_reqwest(
client: &reqwest::Client,
request: http::Request<Vec<u8>>,
max_body_bytes: usize,
) -> Result<http::Response<Vec<u8>>, PublicHttpError> {
let (parts, body) = request.into_parts();
let mut request = client
.request(parts.method, parts.uri.to_string())
.body(body);
for (name, value) in &parts.headers {
request = request.header(name.as_str(), value.as_bytes());
}
let response = request.send().await?;
let status = response.status();
let headers = response.headers().clone();
let mut stream = response.bytes_stream();
let mut body = bytes::BytesMut::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk?;
if body.len() + chunk.len() > max_body_bytes {
return Err(PublicHttpError::BodyTooLarge {
limit: max_body_bytes,
});
}
body.extend_from_slice(&chunk);
}
let mut output = http::Response::builder().status(status);
for (name, value) in &headers {
output = output.header(name, value);
}
Ok(output
.body(body.freeze().to_vec())
.expect("response status and headers are valid"))
}
/// Build the public transport used by legacy XRPC adapters that cannot carry
/// [`PublicEndpoint`] through their trait surface. Callers must validate the
/// endpoint before handing requests to this client; direct CAR callers use the
/// typed `ThrottledHttpClient::get` wrapper.
fn public_http_transport(timeout: Duration, connect_timeout: Duration) -> reqwest::Client {
let connect_timeout = connect_timeout.min(timeout);
reqwest::Client::builder()
.user_agent(concat!(
env!("CARGO_PKG_NAME"),
"/",
env!("CARGO_PKG_VERSION")
))
.dns_resolver(Arc::new(PublicResolver::system()))
.no_proxy()
.redirect(Policy::none())
.timeout(timeout)
.connect_timeout(connect_timeout)
.gzip(true)
.brotli(true)
.zstd(true)
.build()
.expect("public HTTP client configuration should be valid")
}
pub(crate) fn public_http_with_timeout(timeout: Duration) -> PublicHttpClient {
PublicHttpClient {
client: Arc::new(public_http_transport(timeout, PUBLIC_CONNECT_TIMEOUT)),
}
}
/// the same public egress validation as [`public_http`], on the handle budget.
pub(crate) fn handle_http() -> PublicHttpClient {
static CLIENT: OnceLock<PublicHttpClient> = OnceLock::new();
CLIENT
.get_or_init(|| PublicHttpClient {
client: Arc::new(public_http_transport(
HANDLE_REQUEST_TIMEOUT,
HANDLE_CONNECT_TIMEOUT,
)),
})
.clone()
}
/// the same public egress validation as [`public_http`], on the identity budget.
pub(crate) fn identity_http() -> PublicHttpClient {
static CLIENT: OnceLock<PublicHttpClient> = OnceLock::new();
CLIENT
.get_or_init(|| PublicHttpClient {
client: Arc::new(public_http_transport(
IDENTITY_REQUEST_TIMEOUT,
IDENTITY_CONNECT_TIMEOUT,
)),
})
.clone()
}
pub(crate) fn public_http() -> PublicHttpClient {
static CLIENT: OnceLock<PublicHttpClient> = OnceLock::new();
CLIENT
.get_or_init(|| public_http_with_timeout(PUBLIC_REQUEST_TIMEOUT))
.clone()
}
pub(crate) fn trusted_http() -> &'static reqwest::Client {
static CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
CLIENT.get_or_init(|| {
reqwest::Client::builder()
.user_agent(concat!(
env!("CARGO_PKG_NAME"),
"/",
env!("CARGO_PKG_VERSION")
))
.gzip(true)
.brotli(true)
.zstd(true)
// Trust is scoped to the configured origin, not arbitrary redirect
// targets selected by that service.
.redirect(Policy::none())
.build()
.expect("trusted HTTP client configuration should be valid")
})
}
/// Svix's conservative public-address classifier. Keep the canonicalization
/// first: an IPv4-mapped IPv6 literal is an IPv4 address for policy purposes.
fn is_allowed(addr: IpAddr) -> bool {
match addr.to_canonical() {
IpAddr::V4(addr) => {
!(addr.is_private()
|| addr.is_loopback()
|| addr.is_link_local()
|| addr.is_broadcast()
|| addr.is_multicast()
|| addr.is_documentation()
|| is_shared(addr)
|| is_reserved(addr)
|| is_benchmarking(addr)
|| starts_with_zero(addr))
}
IpAddr::V6(addr) => {
!(addr.is_multicast()
|| addr.is_loopback()
|| addr.is_unspecified()
|| is_unicast_link_local(addr)
|| is_site_local(addr)
|| is_unique_local(addr)
|| is_documentation_v6(addr)
|| is_discard_v6(addr)
|| is_iana_non_global_v6(addr)
|| is_6to4_or_nat64_v6(addr)
|| is_srv6(addr)
|| is_transition_or_tunnel_v6(addr))
}
}
}
static ALLOW_PRIVATE_NETWORKS: AtomicBool = AtomicBool::new(false);
/// lets the public egress capability reach non-public addresses, for a local dev stack where
/// every name resolves to loopback. process wide, set from the config when hydrant is created.
pub(crate) fn allow_private_networks(allow: bool) {
ALLOW_PRIVATE_NETWORKS.store(allow, Ordering::Relaxed);
}
/// True for private, reserved, or special-use addresses that must never be
/// reached through the public egress capability.
pub(crate) fn is_private_ip(ip: IpAddr) -> bool {
private_unless_allowed(ip, ALLOW_PRIVATE_NETWORKS.load(Ordering::Relaxed))
}
fn private_unless_allowed(ip: IpAddr, allow_private: bool) -> bool {
!allow_private && !is_allowed(ip)
}
#[inline(always)]
fn is_shared(addr: Ipv4Addr) -> bool {
static CGNAT: Ipv4Net = Ipv4Net::new_assert(Ipv4Addr::new(100, 64, 0, 0), 10);
CGNAT.contains(&addr)
}
#[inline(always)]
fn is_reserved(addr: Ipv4Addr) -> bool {
static IETF_RESERVED_1: Ipv4Net = Ipv4Net::new_assert(Ipv4Addr::new(192, 0, 0, 0), 24);
static IETF_RESERVED_2: Ipv4Net = Ipv4Net::new_assert(Ipv4Addr::new(192, 88, 99, 0), 24);
static CLASS_E: Ipv4Net = Ipv4Net::new_assert(Ipv4Addr::new(240, 0, 0, 0), 4);
// revisit allowing class E addresses if the deployment environment ever
// makes that protocol space routable.
IETF_RESERVED_1.contains(&addr) || IETF_RESERVED_2.contains(&addr) || CLASS_E.contains(&addr)
}
#[inline(always)]
fn is_benchmarking(addr: Ipv4Addr) -> bool {
static BENCHMARKING: Ipv4Net = Ipv4Net::new_assert(Ipv4Addr::new(198, 18, 0, 0), 15);
BENCHMARKING.contains(&addr)
}
#[inline(always)]
fn starts_with_zero(addr: Ipv4Addr) -> bool {
addr.octets()[0] == 0
}
#[inline(always)]
fn is_unicast_link_local(addr: Ipv6Addr) -> bool {
// technically only fe80::/64 are link-local, but fe80::/10 are all reserved
static V6_LL: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0xfe80, 0, 0, 0, 0, 0, 0, 0), 10);
V6_LL.contains(&addr)
}
#[inline(always)]
fn is_site_local(addr: Ipv6Addr) -> bool {
// Deprecated site-local space remains unroutable and is retained as a
// separate rule so future link-local edits cannot accidentally reopen it.
static SITE_LOCAL: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0xfec0, 0, 0, 0, 0, 0, 0, 0), 10);
SITE_LOCAL.contains(&addr)
}
#[inline(always)]
fn is_unique_local(addr: Ipv6Addr) -> bool {
static ULA: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0xfc00, 0, 0, 0, 0, 0, 0, 0), 7);
ULA.contains(&addr)
}
#[inline(always)]
fn is_documentation_v6(addr: Ipv6Addr) -> bool {
static V6_DOCUMENTATION_1: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 0), 32);
static V6_DOCUMENTATION_2: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0x3fff, 0, 0, 0, 0, 0, 0, 0), 20);
V6_DOCUMENTATION_1.contains(&addr) || V6_DOCUMENTATION_2.contains(&addr)
}
#[inline(always)]
fn is_discard_v6(addr: Ipv6Addr) -> bool {
static V6_DISCARD: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0x100, 0, 0, 0, 0, 0, 0, 0), 64);
V6_DISCARD.contains(&addr)
}
#[inline(always)]
fn is_iana_non_global_v6(addr: Ipv6Addr) -> bool {
// IANA special-purpose allocations: benchmarking and ORCHID/ORCHIDv2.
static BENCHMARKING: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0x2, 0, 0, 0, 0, 0, 0), 48);
static ORCHID: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0x10, 0, 0, 0, 0, 0, 0), 28);
static ORCHID_V2: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0x20, 0, 0, 0, 0, 0, 0), 28);
BENCHMARKING.contains(&addr) || ORCHID.contains(&addr) || ORCHID_V2.contains(&addr)
}
#[inline(always)]
fn is_6to4_or_nat64_v6(addr: Ipv6Addr) -> bool {
static IP6TO4: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0x2002, 0, 0, 0, 0, 0, 0, 0), 16);
static NAT64: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0x64, 0xff9b, 0, 0, 0, 0, 0, 0), 96);
static LOCAL_NAT64: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0x64, 0xff9b, 0x1, 0, 0, 0, 0, 0), 48);
IP6TO4.contains(&addr) || NAT64.contains(&addr) || LOCAL_NAT64.contains(&addr)
}
#[inline(always)]
fn is_srv6(addr: Ipv6Addr) -> bool {
static SRV6: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0x5f00, 0, 0, 0, 0, 0, 0, 0), 16);
SRV6.contains(&addr)
}
/// Hydrant defense-in-depth extensions beyond the copied Svix baseline.
/// Deprecated compatibility/translation forms and Teredo can preserve or hide
/// IPv4 routing semantics, so deny each whole prefix rather than decoding it.
#[inline(always)]
fn is_transition_or_tunnel_v6(addr: Ipv6Addr) -> bool {
static IPV4_COMPATIBLE: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0), 96);
static IPV4_TRANSLATED: Ipv6Net =
Ipv6Net::new_assert(Ipv6Addr::new(0, 0, 0, 0, 0xffff, 0, 0, 0), 96);
static TEREDO: Ipv6Net = Ipv6Net::new_assert(Ipv6Addr::new(0x2001, 0, 0, 0, 0, 0, 0, 0), 32);
IPV4_COMPATIBLE.contains(&addr) || IPV4_TRANSLATED.contains(&addr) || TEREDO.contains(&addr)
}
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::*;
use axum::{Router, body::Body, http::StatusCode, response::IntoResponse, routing::get};
use std::sync::Mutex;
use tokio_websockets::{ClientBuilder, ServerBuilder};
#[test]
fn the_handle_budget_is_tighter_than_the_public_one() {
assert!(HANDLE_REQUEST_TIMEOUT < PUBLIC_REQUEST_TIMEOUT);
assert!(IDENTITY_REQUEST_TIMEOUT < PUBLIC_REQUEST_TIMEOUT);
assert!(IDENTITY_CONNECT_TIMEOUT < IDENTITY_REQUEST_TIMEOUT);
assert!(HANDLE_CONNECT_TIMEOUT < HANDLE_REQUEST_TIMEOUT);
assert!(PUBLIC_CONNECT_TIMEOUT < PUBLIC_REQUEST_TIMEOUT);
}
#[test]
fn allowing_private_networks_lets_loopback_through() {
let loopback = IpAddr::from([127, 0, 0, 1]);
let public = IpAddr::from([11, 0, 0, 253]);
assert!(private_unless_allowed(loopback, false));
assert!(!private_unless_allowed(loopback, true));
assert!(!private_unless_allowed(public, false));
}
struct ScriptedLookup(Mutex<Vec<Vec<SocketAddr>>>);
impl Lookup for ScriptedLookup {
fn lookup(
&self,
_host: String,
_port: u16,
) -> Pin<Box<dyn Future<Output = Result<Vec<SocketAddr>, String>> + Send>> {
let result = self.0.lock().unwrap().pop().unwrap_or_default();
Box::pin(std::future::ready(Ok(result)))
}
}
fn resolver(answer: Vec<SocketAddr>) -> PublicResolver {
PublicResolver {
lookup: Arc::new(ScriptedLookup(Mutex::new(vec![answer]))),
}
}
fn addr(ip: &str) -> SocketAddr {
SocketAddr::new(ip.parse().unwrap(), 443)
}
#[tokio::test]
async fn mixed_answers_are_rejected_as_a_set() {
let resolver = resolver(vec![addr("93.184.216.34"), addr("127.0.0.1")]);
assert!(matches!(
resolver.resolve_addresses("pds.example", 443).await,
Err(ResolveError::NonPublic(_))
));
}
#[tokio::test]
async fn websocket_public_resolver_rejects_private_before_tcp() {
let resolver = resolver(vec![addr("127.0.0.1")]);
let uri = "ws://public.example/".parse().unwrap();
let result = ClientBuilder::from_uri(uri)
.resolver(resolver)
.connect()
.await;
assert!(matches!(
result,
Err(tokio_websockets::Error::CannotResolveHost)
));
}
#[derive(Clone)]
struct FixedSocketResolver(SocketAddr);
impl tokio_websockets::resolver::Resolver for FixedSocketResolver {
async fn resolve(
&self,
_host: &str,
_port: u16,
) -> Result<SocketAddr, tokio_websockets::Error> {
Ok(self.0)
}
}
#[tokio::test]
async fn websocket_handshake_retains_original_uri_host() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (tcp, _) = listener.accept().await.unwrap();
let (request, _ws) = ServerBuilder::new().accept(tcp).await.unwrap();
// tokio-websockets exposes the request-target URI as a path, while
// the authority is carried by Host. This is the value TLS/SNI and
// HTTP routing must preserve when the resolver supplies a socket.
assert_eq!(
request.headers().get(http::header::HOST).unwrap(),
"public.example"
);
});
let uri = "ws://public.example/handshake".parse().unwrap();
let (_ws, _response) = ClientBuilder::from_uri(uri)
.resolver(FixedSocketResolver(address))
.connect()
.await
.unwrap();
server.await.unwrap();
}
#[tokio::test]
async fn empty_answers_are_rejected() {
assert!(matches!(
resolver(Vec::new())
.resolve_addresses("pds.example", 443)
.await,
Err(ResolveError::Empty)
));
}
#[tokio::test]
async fn mixed_ipv4_and_ipv6_answers_are_rejected_as_a_set() {
let resolver = resolver(vec![addr("93.184.216.34"), addr("::ffff:127.0.0.1")]);
assert!(matches!(
resolver.resolve_addresses("pds.example", 443).await,
Err(ResolveError::NonPublic(_))
));
}
#[test]
fn svix_is_allowed_regressions() {
// RFC 1918
for address in [
"10.254.0.0",
"192.168.10.65",
"172.16.10.65",
// zero-prefix (including unspecified)
"0.1.2.3",
"0.0.0.0",
// loopback and link-local
"127.0.0.1",
"127.254.254.254",
"169.254.45.1",
// broadcast and multicast
"255.255.255.255",
"224.1.2.3",
"233.252.0.16",
// protocol-reserved, docs, CGNAT, benchmarking, class E
"192.0.0.0",
"192.0.0.255",
"192.0.2.255",
"198.51.100.65",
"203.0.113.6",
"100.100.0.0",
"192.88.99.3",
"198.18.0.0",
"240.240.240.240",
] {
assert!(!is_allowed(address.parse().unwrap()), "accepted {address}");
}
assert!(is_allowed("1.1.1.1".parse().unwrap()));
assert!(is_allowed("8.8.8.8".parse().unwrap()));
for address in [
"::0",
"::1",
"2001:db8::1",
"2001:2::dead:beef",
"2001:10::dead:beef",
"2001:20::dead:beef",
"3fff::dead:beef",
"fe80::dead:beef",
"febf:ffff::dead:beef",
"fec0::dead:beef",
"feff:ffff::dead:beef",
"ff00::dead:beef",
"fdee:3a45:9057::1",
"fd7a:115c:a1e0::1",
// all 6to4 and both reserved NAT64 prefixes are denied.
"2002:c000:0204::dead:beef",
"2002:c0a8:0001::dead:beef",
"64:ff9b::808:808",
"64:ff9b::c0a8:0",
"64:ff9b:1::dead:beef",
// Hydrant defense-in-depth: deprecated compatibility/translation
// and Teredo prefixes are denied without embedded-address decoding.
"::8.8.8.8",
"::ffff:0:8.8.8.8",
"2001:0:ffff:ffff:ffff:ffff:ffff:ffff",
"5f00::dead:beef",
"100::1:2:3:4",
] {
assert!(!is_allowed(address.parse().unwrap()), "accepted {address}");
}
assert!(is_allowed("2001:4860:4860::8888".parse().unwrap()));
assert!(is_allowed("::ffff:8.8.8.8".parse().unwrap()));
assert!(!is_allowed("::ffff:127.0.0.1".parse().unwrap()));
}
#[test]
fn public_endpoint_normalizes_host_and_rejects_unsafe_authority() {
let endpoint =
PublicEndpoint::parse_http(&Url::parse("https://BÃœCHER.Example./xrpc?x=1").unwrap())
.unwrap();
assert_eq!(endpoint.url().host_str(), Some("xn--bcher-kva.example"));
assert_eq!(
endpoint.url().as_str(),
"https://xn--bcher-kva.example/xrpc?x=1"
);
assert!(PublicEndpoint::parse_http(&Url::parse("https://user@example/").unwrap()).is_err());
assert!(
PublicEndpoint::parse_http(&Url::parse("https://example/#fragment").unwrap()).is_err()
);
assert!(PublicEndpoint::parse_http(&Url::parse("https://127.0.0.1/").unwrap()).is_err());
assert!(
PublicEndpoint::parse_http(&Url::parse("https://[::ffff:127.0.0.1]/").unwrap())
.is_err()
);
assert!(
PublicEndpoint::parse_http(&Url::parse("https://[::ffff:8.8.8.8]/").unwrap()).is_ok()
);
assert!(PublicEndpoint::parse_http(&Url::parse("ftp://example/").unwrap()).is_err());
assert!(PublicEndpoint::parse_websocket(&Url::parse("ws://example/").unwrap()).is_err());
assert!(PublicEndpoint::parse_websocket(&Url::parse("wss://example/").unwrap()).is_ok());
assert!(PublicHost::parse("local").is_err());
assert!(PublicHost::parse("example:443").is_err());
for disguised_loopback in ["2130706433", "017700000001", "0x7f000001"] {
assert!(
PublicHost::parse(disguised_loopback).is_err(),
"accepted disguised loopback {disguised_loopback}"
);
}
}
#[tokio::test]
async fn public_client_no_proxy_disables_configured_proxy() {
let app = Router::new().route("/", get(|| async { "direct" }));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = reqwest::Client::builder()
// If no_proxy is removed, this dead local proxy makes the request fail.
.proxy(reqwest::Proxy::all("http://127.0.0.1:9").unwrap())
.no_proxy()
.redirect(Policy::none())
.resolve("public.example", address)
.build()
.unwrap();
let endpoint =
PublicEndpoint::parse_http(&Url::parse("http://public.example/").unwrap()).unwrap();
let response = PublicHttpClient::test(client)
.get(&endpoint)
.send()
.await
.unwrap();
assert_eq!(response.text().await.unwrap(), "direct");
server.abort();
}
#[tokio::test]
async fn public_http_does_not_follow_redirects_to_private_literals() {
let app = Router::new().route(
"/",
get(|| async {
(
StatusCode::FOUND,
[(http::header::LOCATION, "http://127.0.0.1:9/private")],
)
.into_response()
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = reqwest::Client::builder()
.proxy(reqwest::Proxy::all("http://127.0.0.1:9").unwrap())
.no_proxy()
.redirect(Policy::none())
.resolve("public.example", address)
.build()
.unwrap();
let endpoint =
PublicEndpoint::parse_http(&Url::parse("http://public.example/").unwrap()).unwrap();
let response = PublicHttpClient::test(client)
.get(&endpoint)
.send()
.await
.unwrap();
assert_eq!(response.status(), StatusCode::FOUND);
server.abort();
}
#[tokio::test]
async fn public_jacquard_response_body_is_capped_before_materializing() {
let app = Router::new().route(
"/",
get(|| async {
Body::from(vec![b'x'; MAX_PUBLIC_HTTP_BODY_BYTES + 1]).into_response()
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let client = reqwest::Client::builder()
.no_proxy()
.resolve("public.example", address)
.build()
.unwrap();
let request = http::Request::builder()
.uri("http://public.example/")
.body(Vec::new())
.unwrap();
let result = PublicHttpClient::test(client).send_http(request).await;
assert!(matches!(result, Err(PublicHttpError::BodyTooLarge { .. })));
server.abort();
}
}