use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{LazyLock, Mutex};
use std::time::{Duration, Instant};
use anyhow::{bail, Context, Result};
use reqwest::header::{
HeaderName, HeaderValue, AUTHORIZATION, CONTENT_TYPE, COOKIE, PROXY_AUTHORIZATION,
WWW_AUTHENTICATE,
};
use reqwest::{Client, Response};
use url::{Host, Url};
pub const MAX_BODY_BYTES: usize = 8 * 1024 * 1024;
pub(crate) const FETCH_TIMEOUT: Duration = Duration::from_secs(30);
const READ_TIMEOUT: Duration = Duration::from_secs(15);
pub(crate) const MAX_REDIRECTS: usize = 5;
pub(crate) const WORST_CASE_REQUEST: Duration =
Duration::from_secs(FETCH_TIMEOUT.as_secs() * (MAX_REDIRECTS as u64 + 1));
pub fn is_forbidden_ip(ip: &IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => is_forbidden_v4(v4),
IpAddr::V6(v6) => is_forbidden_v6(v6),
}
}
fn is_forbidden_v4(ip: &Ipv4Addr) -> bool {
ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_unspecified() || ip.is_broadcast() || ip.is_multicast() || matches!(ip.octets(), [0, ..])
|| matches!(ip.octets(), [100, b, ..] if (64..=127).contains(&b)) || matches!(ip.octets(), [192, 0, 0, _])
|| matches!(ip.octets(), [198, 18..=19, _, _])
}
fn is_forbidden_v6(ip: &Ipv6Addr) -> bool {
if ip.is_loopback() || ip.is_unspecified() || ip.is_multicast() {
return true;
}
if let Some(v4) = ip.to_ipv4() {
return is_forbidden_v4(&v4);
}
if embedded_v4(ip).iter().any(is_forbidden_v4) {
return true;
}
let seg = ip.segments();
let link_local = (seg[0] & 0xffc0) == 0xfe80;
let ula = (seg[0] & 0xfe00) == 0xfc00;
link_local || ula
}
fn embedded_v4(ip: &Ipv6Addr) -> Vec<Ipv4Addr> {
let seg = ip.segments();
let v4 = |hi: u16, lo: u16| {
Ipv4Addr::new(
(hi >> 8) as u8,
(hi & 0xff) as u8,
(lo >> 8) as u8,
(lo & 0xff) as u8,
)
};
if seg[0] == 0x0064 && seg[1] == 0xff9b {
if seg[2..6] == [0, 0, 0, 0] {
return vec![v4(seg[6], seg[7])];
}
return vec![Ipv4Addr::LOCALHOST];
}
let mut out = Vec::new();
if seg[0] == 0x2002 {
out.push(v4(seg[1], seg[2]));
}
if seg[..4] == [0, 0, 0, 0] && seg[4] == 0xffff && seg[5] == 0 {
out.push(v4(seg[6], seg[7]));
}
if seg[0] == 0x2001 && seg[1] == 0 {
out.push(v4(seg[2], seg[3]));
out.push(v4(seg[6] ^ 0xffff, seg[7] ^ 0xffff));
return out;
}
if seg[5] == 0x5efe {
out.push(v4(seg[6], seg[7]));
}
out
}
fn check_scheme(url: &Url) -> Result<()> {
match url.scheme() {
"http" | "https" => Ok(()),
other => bail!("refusing non-http(s) URL scheme {other:?}"),
}
}
async fn resolve_and_check(url: &Url) -> Result<SocketAddr> {
let host = url.host().context("URL has no host")?;
let port = url
.port_or_known_default()
.context("URL has no usable port")?;
match host {
Host::Ipv4(ip) => {
if is_forbidden_ip(&IpAddr::V4(ip)) {
bail!("refusing to fetch forbidden (internal) address {ip}");
}
Ok(SocketAddr::new(IpAddr::V4(ip), port))
}
Host::Ipv6(ip) => {
if is_forbidden_ip(&IpAddr::V6(ip)) {
bail!("refusing to fetch forbidden (internal) address {ip}");
}
Ok(SocketAddr::new(IpAddr::V6(ip), port))
}
Host::Domain(name) => {
#[cfg(test)]
if let Some(addr) = test_override_for(name, port) {
return Ok(addr);
}
let addrs = tokio::net::lookup_host((name, port))
.await
.with_context(|| format!("resolving host {name:?}"))?;
first_vetted(name, addrs)
}
}
}
fn first_vetted(name: &str, addrs: impl Iterator<Item = SocketAddr>) -> Result<SocketAddr> {
let mut vetted: Option<SocketAddr> = None;
for sa in addrs {
let ip = sa.ip();
if is_forbidden_ip(&ip) {
bail!("refusing to fetch {name:?}: resolves to forbidden address {ip}");
}
if vetted.is_none() {
vetted = Some(sa);
}
}
vetted.ok_or_else(|| anyhow::anyhow!("host {name:?} did not resolve to any address"))
}
#[cfg(test)]
static TEST_HOSTS: std::sync::Mutex<Option<std::collections::HashMap<String, SocketAddr>>> =
std::sync::Mutex::new(None);
#[cfg(test)]
pub(crate) struct TestPki {
pub ca_pem: String,
pub leaf_pem: String,
pub leaf_key_pem: String,
}
#[cfg(test)]
pub(crate) fn test_pki() -> &'static TestPki {
static PKI: std::sync::OnceLock<TestPki> = std::sync::OnceLock::new();
PKI.get_or_init(|| {
use rcgen::{
BasicConstraints, CertificateParams, DnType, IsCa, KeyPair, KeyUsagePurpose, SanType,
};
let mut ca_params = CertificateParams::default();
ca_params
.distinguished_name
.push(DnType::CommonName, "featherreader test CA");
ca_params.is_ca = IsCa::Ca(BasicConstraints::Constrained(0));
ca_params.key_usages = vec![
KeyUsagePurpose::KeyCertSign,
KeyUsagePurpose::CrlSign,
KeyUsagePurpose::DigitalSignature,
];
let ca_key = KeyPair::generate().expect("test CA key");
let ca_cert = ca_params
.clone()
.self_signed(&ca_key)
.expect("test CA cert");
let issuer = rcgen::Issuer::new(ca_params, ca_key);
let mut leaf_params = CertificateParams::default();
leaf_params
.distinguished_name
.push(DnType::CommonName, "featherreader test leaf");
leaf_params.subject_alt_names = TEST_TLS_HOSTS
.iter()
.map(|h| SanType::DnsName((*h).try_into().expect("test SAN")))
.collect();
let leaf_key = KeyPair::generate().expect("test leaf key");
let leaf_cert = leaf_params
.signed_by(&leaf_key, &issuer)
.expect("test leaf cert");
TestPki {
ca_pem: ca_cert.pem(),
leaf_pem: leaf_cert.pem(),
leaf_key_pem: leaf_key.serialize_pem(),
}
})
}
#[cfg(test)]
pub(crate) async fn spawn_tls<F>(
build_routes: F,
) -> (SocketAddr, std::sync::Arc<std::sync::Mutex<Vec<String>>>)
where
F: FnOnce(SocketAddr) -> std::collections::HashMap<String, Vec<TestResponse>>,
{
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer};
static PROVIDER: std::sync::Once = std::sync::Once::new();
PROVIDER.call_once(|| {
let _ = tokio_rustls::rustls::crypto::ring::default_provider().install_default();
});
let pki = test_pki();
let certs: Vec<CertificateDer<'static>> = rustls_pemfile_certs(pki.leaf_pem.as_bytes());
let key: PrivateKeyDer<'static> = rustls_pemfile_key(pki.leaf_key_pem.as_bytes());
let config = tokio_rustls::rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)
.expect("test server TLS config");
let acceptor = tokio_rustls::TlsAcceptor::from(std::sync::Arc::new(config));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let routes = build_routes(addr);
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sink = std::sync::Arc::clone(&log);
let hits: std::sync::Arc<std::sync::Mutex<std::collections::HashMap<String, usize>>> =
Default::default();
tokio::spawn(async move {
loop {
let Ok((sock, _)) = listener.accept().await else {
break;
};
let acceptor = acceptor.clone();
let routes = routes.clone();
let sink = std::sync::Arc::clone(&sink);
let hits = std::sync::Arc::clone(&hits);
tokio::spawn(async move {
let Ok(mut tls) = acceptor.accept(sock).await else {
return;
};
let mut raw: Vec<u8> = Vec::new();
let mut chunk = [0u8; 4096];
loop {
let Ok(n) = tls.read(&mut chunk).await else {
return;
};
if n == 0 {
break;
}
raw.extend_from_slice(&chunk[..n]);
let Some(split) = raw.windows(4).position(|w| w == b"\r\n\r\n") else {
continue;
};
let (head, body) = raw.split_at(split + 4);
let want = String::from_utf8_lossy(head).lines().find_map(|l| {
let (k, v) = l.split_once(':')?;
k.eq_ignore_ascii_case("content-length")
.then(|| v.trim().parse::<usize>().ok())?
});
if want.is_none_or(|want| body.len() >= want) {
break;
}
}
let req = String::from_utf8_lossy(&raw).to_string();
let path = req
.lines()
.next()
.and_then(|l| l.split_whitespace().nth(1))
.unwrap_or("/")
.to_string();
sink.lock().unwrap().push(req);
let n = {
let mut c = hits.lock().unwrap();
let e = c.entry(path.clone()).or_insert(0usize);
let n = *e;
*e += 1;
n
};
let reply = routes
.get(&path)
.and_then(|v| v.get(n.min(v.len().saturating_sub(1))))
.cloned()
.unwrap_or_else(|| TestResponse::json(404, "not found"));
let extra: String = reply
.headers
.iter()
.map(|(k, v)| format!("{k}: {v}\r\n"))
.collect();
let (status, body) = (reply.status, reply.body);
let resp = format!(
"HTTP/1.1 {status} X\r\nContent-Type: application/json\r\n{extra}\
Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = tls.write_all(resp.as_bytes()).await;
let _ = tls.shutdown().await;
});
}
});
(addr, log)
}
#[cfg(test)]
fn rustls_pemfile_certs(
pem: &[u8],
) -> Vec<tokio_rustls::rustls::pki_types::CertificateDer<'static>> {
decode_pem_blocks(pem, "CERTIFICATE")
.into_iter()
.map(Into::into)
.collect()
}
#[cfg(test)]
fn rustls_pemfile_key(pem: &[u8]) -> tokio_rustls::rustls::pki_types::PrivateKeyDer<'static> {
let der = decode_pem_blocks(pem, "PRIVATE KEY")
.into_iter()
.next()
.expect("a private key block");
tokio_rustls::rustls::pki_types::PrivatePkcs8KeyDer::from(der).into()
}
#[cfg(test)]
fn decode_pem_blocks(pem: &[u8], label: &str) -> Vec<Vec<u8>> {
use base64::Engine as _;
let text = String::from_utf8_lossy(pem);
let begin = format!("-----BEGIN {label}-----");
let end = format!("-----END {label}-----");
let mut out = Vec::new();
let mut rest = text.as_ref();
while let Some(i) = rest.find(&begin) {
let after = &rest[i + begin.len()..];
let Some(j) = after.find(&end) else { break };
let b64: String = after[..j].chars().filter(|c| !c.is_whitespace()).collect();
out.push(
base64::engine::general_purpose::STANDARD
.decode(b64)
.expect("valid base64 in test PEM"),
);
rest = &after[j + end.len()..];
}
out
}
#[cfg(test)]
#[derive(Clone)]
pub(crate) struct TestResponse {
pub status: u16,
pub body: String,
pub headers: Vec<(String, String)>,
}
#[cfg(test)]
impl TestResponse {
pub fn json(status: u16, body: impl Into<String>) -> Self {
Self {
status,
body: body.into(),
headers: Vec::new(),
}
}
pub fn with_header(mut self, k: &str, v: &str) -> Self {
self.headers.push((k.to_string(), v.to_string()));
self
}
}
#[cfg(test)]
pub(crate) const TEST_TLS_HOSTS: &[&str] = &[
"pds-e2e.test",
"as-e2e.test",
"feed-tls.test",
"hop-tls.test",
"as-evil.test",
];
#[cfg(test)]
pub(crate) fn test_host_override(host: &str, addr: SocketAddr) {
TEST_HOSTS
.lock()
.unwrap()
.get_or_insert_with(Default::default)
.insert(host.to_string(), addr);
}
#[cfg(test)]
fn test_override_for(name: &str, port: u16) -> Option<SocketAddr> {
let guard = TEST_HOSTS.lock().unwrap();
let map = guard.as_ref()?;
map.get(name)
.copied()
.or_else(|| map.get(&format!("{name}:{port}")).copied())
}
const PINNED_CLIENT_TTL: Duration = Duration::from_secs(300);
const MAX_PINNED_CLIENTS: usize = 256;
const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(60);
struct PinnedClients {
entries: Mutex<HashMap<(String, SocketAddr), (Client, Instant)>>,
builds: AtomicUsize,
}
impl PinnedClients {
fn new() -> Self {
Self {
entries: Mutex::new(HashMap::new()),
builds: AtomicUsize::new(0),
}
}
fn get(&self, host: &str, addr: SocketAddr, now: Instant) -> Result<Client> {
let key = (host.to_string(), addr);
let mut entries = self.entries.lock().unwrap_or_else(|p| p.into_inner());
if let Some((client, last_used)) = entries.get_mut(&key) {
if now.duration_since(*last_used) < PINNED_CLIENT_TTL {
*last_used = now;
return Ok(client.clone());
}
}
let client = build_pinned_client(host, addr)?;
self.builds.fetch_add(1, Ordering::Relaxed);
entries.retain(|_, (_, last_used)| now.duration_since(*last_used) < PINNED_CLIENT_TTL);
if entries.len() >= MAX_PINNED_CLIENTS {
if let Some(oldest) = entries
.iter()
.min_by_key(|(_, (_, last_used))| *last_used)
.map(|(k, _)| k.clone())
{
entries.remove(&oldest);
}
}
entries.insert(key, (client.clone(), now));
Ok(client)
}
}
static PINNED_CLIENTS: LazyLock<PinnedClients> = LazyLock::new(PinnedClients::new);
fn build_pinned_client(host: &str, addr: SocketAddr) -> Result<Client> {
let builder = Client::builder()
.user_agent(crate::USER_AGENT)
.timeout(FETCH_TIMEOUT)
.read_timeout(READ_TIMEOUT)
.pool_max_idle_per_host(1)
.pool_idle_timeout(POOL_IDLE_TIMEOUT)
.no_proxy()
.resolve(host, addr)
.redirect(reqwest::redirect::Policy::none());
#[cfg(test)]
let builder = builder.add_root_certificate(
reqwest::Certificate::from_pem(test_pki().ca_pem.as_bytes())
.context("parsing the test CA")?,
);
builder
.build()
.context("failed to build IP-pinned fetch client")
}
fn pinned_client(host: &str, addr: SocketAddr) -> Result<Client> {
PINNED_CLIENTS.get(host, addr, Instant::now())
}
pub async fn guarded_get(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
) -> Result<Response> {
guarded_get_inner(client, url, extra_headers, true, MAX_REDIRECTS).await
}
pub async fn guarded_get_no_privacy(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
) -> Result<Response> {
guarded_get_inner(client, url, extra_headers, false, MAX_REDIRECTS).await
}
pub async fn guarded_get_no_redirect(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
) -> Result<Response> {
guarded_get_inner(client, url, extra_headers, false, 0).await
}
fn is_sensitive_header(name: &HeaderName) -> bool {
name == AUTHORIZATION
|| name == COOKIE
|| name == PROXY_AUTHORIZATION
|| name == WWW_AUTHENTICATE
|| name.as_str() == "cookie2"
}
fn same_origin(a: &Url, b: &Url) -> bool {
a.scheme() == b.scheme()
&& a.host_str() == b.host_str()
&& a.port_or_known_default() == b.port_or_known_default()
}
fn hop_headers<'a>(
original: &Url,
current: &Url,
extra: &'a [(HeaderName, HeaderValue)],
) -> Vec<&'a (HeaderName, HeaderValue)> {
let cross_origin = !same_origin(original, current);
extra
.iter()
.filter(|(name, _)| !(cross_origin && is_sensitive_header(name)))
.collect()
}
async fn guarded_get_inner(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
check_privacy: bool,
max_redirects: usize,
) -> Result<Response> {
let _ = client;
let mut current = Url::parse(url).with_context(|| format!("not a valid URL {url:?}"))?;
let original = current.clone();
for _ in 0..=max_redirects {
check_scheme(¤t)?;
if check_privacy {
if let crate::feed::FeedPrivacy::Private(reason) =
crate::feed::classify_feed_privacy(current.as_str())
{
bail!("refusing to fetch private/paid feed URL (redirect target): {reason}");
}
}
let vetted = resolve_and_check(¤t).await?;
let host = current
.host_str()
.context("URL lost its host between hops")?
.to_string();
let hop_client = pinned_client(&host, vetted)?;
let mut req = hop_client.get(current.clone());
for (name, value) in hop_headers(&original, ¤t, extra_headers) {
req = req.header(name.clone(), value.clone());
}
let resp = req
.send()
.await
.with_context(|| format!("fetching {current}"))?;
if resp.status() != reqwest::StatusCode::NOT_MODIFIED && resp.status().is_redirection() {
if max_redirects == 0 {
bail!(
"refusing to follow a {} redirect while fetching {url:?} \u{2014} \
this document's origin is load-bearing and must not be moved",
resp.status()
);
}
if !matches!(resp.status().as_u16(), 301 | 302 | 303 | 307 | 308) {
bail!(
"refusing to act on a {} response while fetching {url:?} \u{2014} \
it names no single target that can be followed safely",
resp.status()
);
}
let location = resp
.headers()
.get(reqwest::header::LOCATION)
.and_then(|v| v.to_str().ok())
.context("redirect response without a usable Location header")?;
current = current
.join(location)
.with_context(|| format!("resolving redirect Location {location:?}"))?;
continue;
}
return Ok(resp);
}
bail!("too many redirects (> {max_redirects}) while fetching {url:?}")
}
pub async fn guarded_post_json(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
body: Vec<u8>,
) -> Result<Response> {
guarded_post(client, url, extra_headers, PostBody::Json(body)).await
}
pub(crate) enum PostBody<'a> {
Json(Vec<u8>),
Form(&'a [(&'a str, &'a str)]),
}
impl PostBody<'_> {
fn content_type(&self) -> HeaderValue {
match self {
PostBody::Json(_) => HeaderValue::from_static("application/json"),
PostBody::Form(_) => HeaderValue::from_static("application/x-www-form-urlencoded"),
}
}
fn into_bytes(self) -> Vec<u8> {
match self {
PostBody::Json(bytes) => bytes,
PostBody::Form(params) => {
let mut ser = url::form_urlencoded::Serializer::new(String::new());
for (k, v) in params {
ser.append_pair(k, v);
}
ser.finish().into_bytes()
}
}
}
}
pub async fn guarded_post_form(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
params: &[(&str, &str)],
) -> Result<Response> {
guarded_post(client, url, extra_headers, PostBody::Form(params)).await
}
async fn guarded_post(
client: &Client,
url: &str,
extra_headers: &[(HeaderName, HeaderValue)],
body: PostBody<'_>,
) -> Result<Response> {
let content_type = body.content_type();
let body = body.into_bytes();
let _ = client;
let target = Url::parse(url).with_context(|| format!("not a valid URL {url:?}"))?;
check_scheme(&target)?;
let vetted = resolve_and_check(&target).await?;
let host = target.host_str().context("URL has no host")?.to_string();
let hop_client = pinned_client(&host, vetted)?;
let mut req = hop_client
.post(target.clone())
.header(CONTENT_TYPE, content_type)
.body(body);
for (name, value) in extra_headers {
req = req.header(name.clone(), value.clone());
}
let resp = req
.send()
.await
.with_context(|| format!("posting to {target}"))?;
if resp.status().is_redirection() {
if resp.status() == reqwest::StatusCode::NOT_MODIFIED {
bail!(
"a POST to {url:?} answered 304 Not Modified, which is not a valid \
response to a request carrying no conditional headers"
);
}
let location = resp
.headers()
.get(reqwest::header::LOCATION)
.and_then(|v| v.to_str().ok())
.unwrap_or("<none>");
bail!(
"refusing to follow a {} redirect on a POST to {url:?} (Location: {location}) — \
a 307/308 would re-send the request body to the new host",
resp.status()
);
}
Ok(resp)
}
pub async fn read_capped(mut resp: Response) -> Result<Vec<u8>> {
let mut buf: Vec<u8> = Vec::with_capacity(16 * 1024);
while let Some(chunk) = resp.chunk().await.context("reading response body chunk")? {
if buf.len() + chunk.len() > MAX_BODY_BYTES {
bail!(
"response body exceeded the {} byte cap; aborting",
MAX_BODY_BYTES
);
}
buf.extend_from_slice(&chunk);
}
Ok(buf)
}
pub async fn assert_public_target(url: &str) -> Result<()> {
let parsed = Url::parse(url).with_context(|| format!("not a valid URL {url:?}"))?;
check_scheme(&parsed)?;
resolve_and_check(&parsed).await?;
Ok(())
}
pub fn safe_link(raw: &str) -> Option<String> {
let trimmed = raw.trim();
if trimmed.is_empty() {
return None;
}
match Url::parse(trimmed) {
Ok(u) if matches!(u.scheme(), "http" | "https") => Some(trimmed.to_string()),
_ => None,
}
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
#[test]
fn forbids_loopback_and_link_local_and_private_v4() {
for ip in [
"127.0.0.1",
"127.1.2.3",
"169.254.169.254", "10.0.0.5",
"172.16.9.9",
"192.168.1.1",
"0.0.0.0",
"255.255.255.255",
"100.64.0.1", ] {
let ip: IpAddr = ip.parse().unwrap();
assert!(is_forbidden_ip(&ip), "{ip} should be forbidden");
}
}
#[test]
fn allows_public_v4() {
for ip in ["1.1.1.1", "8.8.8.8", "93.184.216.34"] {
let ip: IpAddr = ip.parse().unwrap();
assert!(!is_forbidden_ip(&ip), "{ip} should be allowed");
}
}
#[test]
fn forbids_internal_v6() {
for ip in [
"::1",
"fe80::1",
"fc00::1",
"fd00::1",
"::ffff:127.0.0.1",
"::",
] {
let ip: IpAddr = ip.parse().unwrap();
assert!(is_forbidden_ip(&ip), "{ip} should be forbidden");
}
}
#[test]
fn forbids_ipv6_that_embeds_a_forbidden_ipv4() {
for (ip, what) in [
("64:ff9b::7f00:1", "NAT64 well-known -> 127.0.0.1"),
("64:ff9b::a9fe:a9fe", "NAT64 well-known -> 169.254.169.254"),
("64:ff9b::c0a8:1", "NAT64 well-known -> 192.168.0.1"),
("64:ff9b:1::7f00:1", "NAT64 local-use, RFC 8215"),
("64:ff9b:1:ffff::1", "anywhere in the NAT64 /32"),
("2002:7f00:1::", "6to4 -> 127.0.0.1"),
("2002:a9fe:a9fe::", "6to4 -> 169.254.169.254"),
("::ffff:0:7f00:1", "IPv4-translated -> 127.0.0.1"),
("2001:0:7f00:1:0:0:f7f7:fbfb", "Teredo server -> 127.0.0.1"),
("2001:0:808:808:0:0:80ff:fffe", "Teredo client -> 127.0.0.1"),
(
"2001:0:808:808:0:0:5601:5601",
"Teredo client -> 169.254.169.254",
),
("2001:db8::5efe:7f00:1", "ISATAP -> 127.0.0.1"),
("2001:db8::5efe:a9fe:a9fe", "ISATAP -> 169.254.169.254"),
(
"2001:db8::200:5efe:a9fe:a9fe",
"ISATAP u=1 g=0 -> 169.254.169.254",
),
("2001:db8::100:5efe:7f00:1", "ISATAP u=0 g=1 -> 127.0.0.1"),
("2001:db8::300:5efe:7f00:1", "ISATAP u=1 g=1 -> 127.0.0.1"),
(
"2606:4700::5efe:c0a8:1",
"ISATAP under a REAL public prefix -> 192.168.0.1",
),
(
"2002:808:808:0:0:5efe:a9fe:a9fe",
"6to4 site 8.8.8.8 + ISATAP -> 169.254.169.254",
),
(
"2002:808:808:0:0:5efe:7f00:1",
"6to4 site 8.8.8.8 + ISATAP -> 127.0.0.1",
),
(
"2002:101:101:0:0:5efe:c0a8:1",
"6to4 site 1.1.1.1 + ISATAP -> 192.168.0.1",
),
] {
let parsed: IpAddr = ip.parse().unwrap();
assert!(
is_forbidden_ip(&parsed),
"{ip} reaches {what} and was allowed",
);
}
}
#[test]
fn allows_ipv6_that_embeds_a_public_ipv4() {
for (ip, what) in [
("2002:0808:0808::", "6to4 -> 8.8.8.8"),
(
"2001:0:808:808:0:0:f7f7:fbfb",
"Teredo, server 8.8.8.8 and client 8.8.4.4",
),
("::ffff:0:808:808", "IPv4-translated -> 8.8.8.8"),
("2606:4700::5efe:808:808", "ISATAP -> 8.8.8.8"),
("64:ff9b::808:808", "NAT64 well-known prefix -> 8.8.8.8"),
(
"2002:808:808:0:0:5efe:808:404",
"6to4 site 8.8.8.8 + ISATAP 8.8.4.4",
),
] {
let parsed: IpAddr = ip.parse().unwrap();
assert!(!is_forbidden_ip(&parsed), "{ip} is {what} and was refused");
}
}
#[test]
fn a_local_use_nat64_prefix_is_refused_even_wrapping_a_public_address() {
let ip: IpAddr = "64:ff9b:1::808:808".parse().unwrap();
assert!(
is_forbidden_ip(&ip),
"the local-use NAT64 prefix was decoded as if its embedding length \
were known",
);
}
#[test]
fn a_dns64_answer_set_for_an_ipv4_only_host_is_fetchable() {
let synthesised: SocketAddr = "[64:ff9b::808:808]:80".parse().unwrap();
let public_v4: SocketAddr = "1.2.3.4:80".parse().unwrap();
assert_eq!(
first_vetted("v4only.example", [synthesised].into_iter()).unwrap(),
synthesised,
"a DNS64-synthesised answer for a public host was refused, which \
makes every IPv4-only publisher unfetchable behind NAT64",
);
assert!(first_vetted("both.example", [public_v4, synthesised].into_iter()).is_ok());
let hostile: SocketAddr = "[64:ff9b::a9fe:a9fe]:80".parse().unwrap();
assert!(
first_vetted("evil.example", [public_v4, hostile].into_iter()).is_err(),
"a NAT64-synthesised metadata address was accepted",
);
assert!(first_vetted("evil.example", [hostile].into_iter()).is_err());
}
#[test]
fn a_reserved_bit_in_the_isatap_identifier_does_not_buy_a_bypass() {
let ip: IpAddr = "2001:db8::400:5efe:7f00:1".parse().unwrap();
assert!(
is_forbidden_ip(&ip),
"an identifier carrying 00-00-5E-FE and 127.0.0.1 was allowed \
because a reserved bit was set",
);
}
#[test]
fn a_teredo_address_is_read_as_teredo_not_as_isatap() {
let ip: IpAddr = "2001:0:808:808:0:5efe:7f00:1".parse().unwrap();
assert!(
!is_forbidden_ip(&ip),
"an address in Teredo space was read as ISATAP and wrongly refused",
);
}
#[test]
fn a_link_local_isatap_address_is_still_refused_for_being_link_local() {
let ip: IpAddr = "fe80::5efe:808:808".parse().unwrap();
assert!(
is_forbidden_ip(&ip),
"fe80::/10 wrapping a public IPv4 escaped the link-local rule",
);
}
#[test]
fn allows_public_v6() {
let ip: IpAddr = "2606:4700:4700::1111".parse().unwrap();
assert!(!is_forbidden_ip(&ip));
}
#[tokio::test]
async fn resolve_and_check_rejects_ip_literals() {
for bad in [
"http://127.0.0.1/feed.xml",
"http://169.254.169.254/latest/meta-data/",
"http://[::1]:80/x",
"http://192.168.0.1/",
] {
let u = Url::parse(bad).unwrap();
assert!(
resolve_and_check(&u).await.is_err(),
"{bad} should be rejected"
);
}
}
#[tokio::test]
async fn resolve_and_check_allows_public_ip_literal() {
let u = Url::parse("http://1.1.1.1/").unwrap();
let addr = resolve_and_check(&u).await.unwrap();
assert_eq!(addr, "1.1.1.1:80".parse::<SocketAddr>().unwrap());
}
#[tokio::test]
async fn resolve_and_check_pins_public_ipv6_literal() {
let u = Url::parse("http://[2606:4700:4700::1111]:443/").unwrap();
let addr = resolve_and_check(&u).await.unwrap();
assert_eq!(
addr,
"[2606:4700:4700::1111]:443".parse::<SocketAddr>().unwrap()
);
}
const V4: &str = "93.184.216.34:443";
const V4_OTHER: &str = "93.184.216.35:443";
fn at(base: Instant, secs: u64) -> Instant {
base + Duration::from_secs(secs)
}
#[test]
fn the_same_vetted_address_reuses_one_client() {
let cache = PinnedClients::new();
let now = Instant::now();
let addr: SocketAddr = V4.parse().unwrap();
for i in 0..5 {
cache.get("example.com", addr, at(now, i)).unwrap();
}
assert_eq!(
cache.builds.load(Ordering::Relaxed),
1,
"each request rebuilt the client, so every call pays a TLS handshake"
);
}
#[test]
fn a_changed_address_does_not_reuse_the_pooled_client() {
let cache = PinnedClients::new();
let now = Instant::now();
cache.get("example.com", V4.parse().unwrap(), now).unwrap();
cache
.get("example.com", V4_OTHER.parse().unwrap(), at(now, 1))
.unwrap();
assert_eq!(
cache.builds.load(Ordering::Relaxed),
2,
"the same host at a DIFFERENT address reused a connection pinned to the old one"
);
assert_eq!(cache.entries.lock().unwrap().len(), 2);
}
#[test]
fn different_hosts_at_one_address_are_separate_clients() {
let cache = PinnedClients::new();
let now = Instant::now();
let addr: SocketAddr = V4.parse().unwrap();
cache.get("a.example.com", addr, now).unwrap();
cache.get("b.example.com", addr, now).unwrap();
assert_eq!(cache.builds.load(Ordering::Relaxed), 2);
}
#[test]
fn an_idle_entry_is_rebuilt_after_the_ttl() {
let cache = PinnedClients::new();
let now = Instant::now();
let addr: SocketAddr = V4.parse().unwrap();
cache.get("example.com", addr, now).unwrap();
cache
.get(
"example.com",
addr,
now + PINNED_CLIENT_TTL + Duration::from_secs(1),
)
.unwrap();
assert_eq!(cache.builds.load(Ordering::Relaxed), 2);
}
#[test]
fn continued_use_keeps_an_entry_alive() {
let cache = PinnedClients::new();
let now = Instant::now();
let addr: SocketAddr = V4.parse().unwrap();
for minute in 0..20 {
cache
.get("example.com", addr, at(now, minute * 60))
.unwrap();
}
assert_eq!(
cache.builds.load(Ordering::Relaxed),
1,
"a continuously-used client was expired by age rather than idleness"
);
}
#[test]
fn idle_sockets_are_released_before_their_client_is() {
assert!(
POOL_IDLE_TIMEOUT < PINNED_CLIENT_TTL,
"pool idle timeout {POOL_IDLE_TIMEOUT:?} is not shorter than the \
client TTL {PINNED_CLIENT_TTL:?}"
);
}
#[test]
fn the_cache_is_bounded() {
let cache = PinnedClients::new();
let now = Instant::now();
for i in 0..(MAX_PINNED_CLIENTS + 50) {
let addr: SocketAddr = format!("93.184.216.34:{}", 1024 + i).parse().unwrap();
cache.get(&format!("h{i}.example.com"), addr, now).unwrap();
}
assert!(
cache.entries.lock().unwrap().len() <= MAX_PINNED_CLIENTS,
"the cache grew past its bound"
);
}
#[test]
fn pinned_client_builds_for_both_families() {
assert!(pinned_client("example.com", "93.184.216.34:80".parse().unwrap()).is_ok());
assert!(
pinned_client("example.com", "[2606:4700:4700::1111]:443".parse().unwrap()).is_ok()
);
}
#[test]
fn scheme_allowlist_rejects_non_http() {
assert!(check_scheme(&Url::parse("http://example.com/").unwrap()).is_ok());
assert!(check_scheme(&Url::parse("https://example.com/").unwrap()).is_ok());
assert!(check_scheme(&Url::parse("file:///etc/passwd").unwrap()).is_err());
assert!(check_scheme(&Url::parse("ftp://example.com/").unwrap()).is_err());
}
pub(crate) async fn serve_json_capturing(
reply: Vec<u8>,
) -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sink = std::sync::Arc::clone(&log);
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = listener.accept().await else {
break;
};
let mut raw: Vec<u8> = Vec::new();
let mut chunk = [0u8; 4096];
let text = loop {
let Ok(n) = sock.read(&mut chunk).await else {
break String::new();
};
if n == 0 {
break String::from_utf8_lossy(&raw).to_string();
}
raw.extend_from_slice(&chunk[..n]);
let Some(split) = raw.windows(4).position(|w| w == b"\r\n\r\n") else {
continue;
};
let (head, body) = raw.split_at(split + 4);
let want = String::from_utf8_lossy(head).lines().find_map(|l| {
let (k, v) = l.split_once(':')?;
k.eq_ignore_ascii_case("content-length")
.then(|| v.trim().parse::<usize>().ok())?
});
if want.is_none_or(|w| body.len() >= w) {
break String::from_utf8_lossy(&raw).to_string();
}
};
sink.lock().unwrap().push(text);
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
reply.len()
);
let _ = sock.write_all(header.as_bytes()).await;
let _ = sock.write_all(&reply).await;
let _ = sock.flush().await;
}
});
(format!("http://{addr}"), log)
}
pub(crate) async fn serve_body_counted(
body: Vec<u8>,
) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let hits = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let counter = std::sync::Arc::clone(&hits);
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 sock, _)) = listener.accept().await else {
break;
};
counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let body = body.clone();
tokio::spawn(async move {
let mut buf = [0u8; 1024];
let _ = sock.read(&mut buf).await;
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/rss+xml\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = sock.write_all(header.as_bytes()).await;
let _ = sock.write_all(&body).await;
let _ = sock.flush().await;
});
}
});
(format!("http://{addr}/"), hits)
}
pub(crate) async fn serve_bodies_in_sequence(bodies: Vec<Vec<u8>>) -> String {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
assert!(!bodies.is_empty(), "serve_bodies_in_sequence needs a body");
let bodies = std::sync::Arc::new(bodies);
let next = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
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 sock, _)) = listener.accept().await else {
break;
};
let i = next.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let body = bodies[i.min(bodies.len() - 1)].clone();
tokio::spawn(async move {
let mut req = Vec::new();
let mut buf = [0u8; 1024];
let mut want: Option<usize> = None;
loop {
match sock.read(&mut buf).await {
Ok(0) => break,
Ok(n) => {
req.extend_from_slice(&buf[..n]);
let Some(head_end) = req.windows(4).position(|w| w == b"\r\n\r\n")
else {
continue;
};
let head_len = head_end + 4;
if want.is_none() {
let head = String::from_utf8_lossy(&req[..head_len]);
want = Some(
head.lines()
.find_map(|l| {
let (k, v) = l.split_once(':')?;
k.eq_ignore_ascii_case("content-length")
.then(|| v.trim().parse::<usize>().ok())?
})
.unwrap_or(0),
);
}
if req.len() >= head_len + want.unwrap_or(0) {
break;
}
}
Err(_) => break,
}
}
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = sock.write_all(header.as_bytes()).await;
let _ = sock.write_all(&body).await;
let _ = sock.flush().await;
});
}
});
format!("http://{addr}/")
}
pub(crate) async fn serve_body(body: Vec<u8>) -> String {
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 (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => break,
};
let body = body.clone();
tokio::spawn(async move {
let mut buf = [0u8; 1024];
let _ = sock.read(&mut buf).await;
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = sock.write_all(header.as_bytes()).await;
let _ = sock.write_all(&body).await;
let _ = sock.flush().await;
});
}
});
format!("http://{addr}/")
}
#[tokio::test]
async fn read_capped_rejects_over_cap_body() {
let big = vec![b'x'; MAX_BODY_BYTES + 1];
let base = serve_body(big).await;
let client = reqwest::Client::builder().build().unwrap();
let resp = client.get(&base).send().await.unwrap();
let err = read_capped(resp).await.unwrap_err().to_string();
assert!(err.contains("exceeded"), "unexpected error: {err}");
}
#[tokio::test]
async fn read_capped_accepts_small_body() {
let base = serve_body(b"hello world".to_vec()).await;
let client = reqwest::Client::builder().build().unwrap();
let resp = client.get(&base).send().await.unwrap();
let body = read_capped(resp).await.unwrap();
assert_eq!(body, b"hello world");
}
pub(crate) async fn serve_redirect_to_final() -> 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 (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => break,
};
tokio::spawn(async move {
let mut buf = [0u8; 1024];
let n = sock.read(&mut buf).await.unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]).to_string();
let resp = if req.starts_with("GET /final") {
"HTTP/1.1 200 OK\r\nContent-Length: 7\r\nConnection: close\r\n\r\narrived"
} else {
"HTTP/1.1 302 Found\r\nLocation: /final\r\nContent-Length: 0\r\n\
Connection: close\r\n\r\n"
};
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.flush().await;
});
}
});
addr
}
#[tokio::test]
async fn the_connect_is_pinned_to_the_vetted_address() {
let addr = serve_redirect_to_final().await;
let host = "pinned-target.invalid";
let client = pinned_client(host, addr).expect("building a pinned client");
let resp = client
.get(format!("http://{host}:{}/final", addr.port()))
.send()
.await
.expect(
"a pinned host must reach the vetted address without consulting DNS — \
if this failed to connect, the `.resolve()` pin is gone",
);
assert_eq!(resp.status(), 200);
assert_eq!(resp.text().await.unwrap(), "arrived");
}
#[tokio::test]
async fn the_pinned_client_does_not_follow_redirects_itself() {
let addr = serve_redirect_to_final().await;
let host = "redirector.invalid";
let client = pinned_client(host, addr).expect("building a pinned client");
let resp = client
.get(format!("http://{host}:{}/start", addr.port()))
.send()
.await
.expect("the stub must answer the first hop");
assert_eq!(
resp.status(),
302,
"the per-hop client must hand the 30x BACK to guarded_get_inner for \
re-validation; a 200 here means reqwest followed it internally and the \
second hop was connected to without passing resolve_and_check",
);
assert_eq!(
resp.headers()
.get(reqwest::header::LOCATION)
.and_then(|v| v.to_str().ok()),
Some("/final"),
"the Location must reach the caller — it is what the next hop re-validates",
);
}
#[test]
fn every_blocklist_branch_is_load_bearing() {
for ip in [
"224.0.0.1", "239.255.255.250", "192.0.0.1", "192.0.0.171", "198.18.0.1", "198.19.255.255", "0.0.0.0", "0.1.2.3", "255.255.255.255", "100.64.0.1", "100.127.255.255", ] {
let parsed: IpAddr = ip.parse().unwrap();
assert!(is_forbidden_ip(&parsed), "{ip} must be forbidden");
}
for ip in [
"ff02::1", "::1", "::", "fe80::1", "fc00::1", "fd00::1", "::ffff:127.0.0.1", "::ffff:169.254.169.254", "::ffff:10.0.0.1", ] {
let parsed: IpAddr = ip.parse().unwrap();
assert!(is_forbidden_ip(&parsed), "{ip} must be forbidden");
}
}
#[test]
fn the_blocklist_does_not_over_block_adjacent_public_space() {
for ip in [
"9.255.255.255", "11.0.0.0", "172.15.255.255", "172.32.0.0", "192.167.255.255", "192.169.0.0", "169.253.255.255", "169.255.0.0", "126.255.255.255", "128.0.0.0", "100.63.255.255", "100.128.0.0", "192.0.1.0", "198.17.255.255", "198.20.0.0", "223.255.255.255", "1.0.0.0", ] {
let parsed: IpAddr = ip.parse().unwrap();
assert!(
!is_forbidden_ip(&parsed),
"{ip} is public and adjacent to a blocked range — refusing it means a \
CIDR boundary is wrong and real feeds are unreachable",
);
}
for ip in ["2606:4700:4700::1111", "2001:4860:4860::8888"] {
let parsed: IpAddr = ip.parse().unwrap();
assert!(
!is_forbidden_ip(&parsed),
"{ip} is public and must be allowed"
);
}
}
#[tokio::test]
async fn guarded_post_refuses_internal_targets() {
let client = Client::builder().build().unwrap();
for url in [
"http://127.0.0.1:9/xrpc/com.atproto.server.createSession",
"http://169.254.169.254/latest/meta-data/",
"http://10.0.0.5/xrpc/com.atproto.repo.applyWrites",
"http://[::1]/xrpc/com.atproto.repo.deleteRecord",
] {
let err = guarded_post_json(&client, url, &[], b"{}".to_vec())
.await
.unwrap_err()
.to_string();
assert!(
err.contains("forbidden") || err.contains("internal"),
"{url}: expected an SSRF refusal, got: {err}"
);
}
}
#[tokio::test]
async fn guarded_post_refuses_bad_schemes() {
let client = Client::builder().build().unwrap();
for url in ["file:///etc/passwd", "gopher://example.com/1"] {
let err = guarded_post_json(&client, url, &[], b"{}".to_vec())
.await
.unwrap_err()
.to_string();
assert!(err.contains("scheme"), "{url}: got: {err}");
}
}
#[tokio::test]
async fn a_redirect_to_a_private_feed_is_refused_before_it_is_fetched() {
let (target_addr, target_log) = spawn_http(vec![ok_200()]).await;
test_host_override("private-target.test", target_addr);
let (hop_addr, hop_log) = spawn_http(vec![redirect_to(&format!(
"http://private-target.test:{}/feed/private/deadbeefcafe1234",
target_addr.port()
))])
.await;
test_host_override("private-hop.test", hop_addr);
let err = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://private-hop.test:{}/feed.xml", hop_addr.port()),
&[],
)
.await
.expect_err("a redirect to a private feed was followed");
let rendered = format!("{err:#}");
assert!(
rendered.contains("private/paid feed URL (redirect target)"),
"refused for the wrong reason: {rendered}"
);
assert_eq!(
hop_log.lock().unwrap().len(),
1,
"the public first hop is fetched"
);
assert_eq!(
target_log.lock().unwrap().len(),
0,
"the private target was FETCHED before being refused"
);
}
fn hop_fixture() -> Vec<(HeaderName, HeaderValue)> {
vec![
(AUTHORIZATION, HeaderValue::from_static("Bearer secret")),
(
reqwest::header::IF_NONE_MATCH,
HeaderValue::from_static("\"etag\""),
),
]
}
#[test]
fn sensitive_header_set() {
assert!(is_sensitive_header(&AUTHORIZATION));
assert!(is_sensitive_header(&COOKIE));
assert!(is_sensitive_header(&PROXY_AUTHORIZATION));
assert!(is_sensitive_header(&WWW_AUTHENTICATE));
assert!(is_sensitive_header(&HeaderName::from_static("cookie2")));
assert!(!is_sensitive_header(&reqwest::header::IF_NONE_MATCH));
assert!(!is_sensitive_header(&reqwest::header::IF_MODIFIED_SINCE));
assert!(!is_sensitive_header(&reqwest::header::ACCEPT));
}
#[test]
fn hop_headers_keeps_all_on_same_origin() {
let extra = hop_fixture();
let original = Url::parse("https://pds.example.com/xrpc/x").unwrap();
assert_eq!(hop_headers(&original, &original, &extra).len(), 2);
let same = Url::parse("https://pds.example.com/other/path?q=1").unwrap();
assert_eq!(hop_headers(&original, &same, &extra).len(), 2);
}
#[test]
fn hop_headers_strips_authorization_cross_host() {
let extra = hop_fixture();
let original = Url::parse("https://pds.example.com/x").unwrap();
let evil = Url::parse("https://evil.example.net/y").unwrap();
let kept = hop_headers(&original, &evil, &extra);
assert_eq!(kept.len(), 1, "the bearer must not follow a cross-host 302");
assert_eq!(kept[0].0, reqwest::header::IF_NONE_MATCH);
}
#[test]
fn hop_headers_strips_on_port_and_scheme_change() {
let extra = hop_fixture();
let original = Url::parse("https://a.example/x").unwrap();
for downgraded in ["http://a.example/x", "https://a.example:8443/x"] {
let current = Url::parse(downgraded).unwrap();
let kept = hop_headers(&original, ¤t, &extra);
assert_eq!(kept.len(), 1, "{downgraded} must drop the credential");
assert_eq!(kept[0].0, reqwest::header::IF_NONE_MATCH);
}
let explicit = Url::parse("https://a.example:443/x").unwrap();
assert_eq!(hop_headers(&original, &explicit, &extra).len(), 2);
}
#[test]
fn safe_link_allowlist() {
assert_eq!(
safe_link("https://ok.example/x").as_deref(),
Some("https://ok.example/x")
);
assert_eq!(
safe_link(" http://ok.example/ ").as_deref(),
Some("http://ok.example/")
);
assert_eq!(safe_link("javascript:alert(document.domain)"), None);
assert_eq!(safe_link("data:text/html,<script>alert(1)</script>"), None);
assert_eq!(safe_link(""), None);
assert_eq!(safe_link(" "), None);
assert_eq!(safe_link("/relative/path"), None);
}
#[tokio::test]
async fn guarded_post_form_fails_closed_on_a_forbidden_target() {
let client = Client::new();
for url in [
"http://127.0.0.1:2583/oauth/token",
"http://[::1]:2583/oauth/token",
"http://169.254.169.254/latest/meta-data/",
"http://10.0.0.5/oauth/token",
] {
let err = guarded_post_form(&client, url, &[], &[("grant_type", "authorization_code")])
.await
.expect_err("must refuse {url}");
let msg = err.to_string().to_lowercase();
assert!(
msg.contains("forbidden") || msg.contains("refus") || msg.contains("resolve"),
"unexpected error for {url}: {err:#}"
);
}
}
#[tokio::test]
async fn guarded_post_form_rejects_non_http_schemes() {
let client = Client::new();
assert!(
guarded_post_form(&client, "file:///etc/passwd", &[], &[("a", "b")])
.await
.is_err()
);
}
#[tokio::test]
async fn guarded_get_no_redirect_still_fails_closed_on_forbidden_targets() {
let client = Client::new();
for url in [
"http://127.0.0.1/.well-known/oauth-authorization-server",
"http://169.254.169.254/latest/meta-data/",
"http://192.168.1.1/.well-known/did.json",
"http://[::1]/.well-known/did.json",
] {
let err = guarded_get_no_redirect(&client, url, &[])
.await
.expect_err("must refuse");
let rendered = format!("{err:#}");
assert!(
rendered.contains("forbidden (internal) address"),
"{url} failed for the wrong reason: {rendered}"
);
}
let err = guarded_get_no_redirect(&client, "file:///etc/passwd", &[])
.await
.expect_err("must refuse");
assert!(format!("{err:#}").contains("non-http(s) URL scheme"));
}
#[tokio::test]
async fn guarded_get_no_redirect_refuses_to_follow_even_one_hop() {
let (b_addr, b_log) = spawn_http(vec![ok_200()]).await;
test_host_override("no-redirect-b.test", b_addr);
let (a_addr, a_log) = spawn_http(vec![redirect_to(&format!(
"http://no-redirect-b.test:{}/.well-known/oauth-authorization-server",
b_addr.port()
))])
.await;
test_host_override("no-redirect-a.test", a_addr);
let err = guarded_get_no_redirect(
&reqwest::Client::builder().build().unwrap(),
&format!(
"http://no-redirect-a.test:{}/.well-known/oauth-authorization-server",
a_addr.port()
),
&[],
)
.await
.expect_err("a redirect was followed on the no-redirect path");
let rendered = format!("{err:#}");
assert!(
rendered.contains("origin is load-bearing"),
"refused for the wrong reason: {rendered}"
);
assert_eq!(a_log.lock().unwrap().len(), 1);
assert_eq!(
b_log.lock().unwrap().len(),
0,
"the redirect target was fetched — the hop was followed"
);
}
#[tokio::test]
async fn guarded_post_refuses_a_redirect_rather_than_resending_the_body() {
let (elsewhere_addr, elsewhere_log) = spawn_http(vec![ok_200()]).await;
test_host_override("post-elsewhere.test", elsewhere_addr);
let (addr, log) = spawn_http(vec![format!(
"HTTP/1.1 307 Temporary Redirect\r\nLocation: http://post-elsewhere.test:{}/token\r\n\
Content-Length: 0\r\nConnection: close\r\n\r\n",
elsewhere_addr.port()
)])
.await;
test_host_override("post-redirect.test", addr);
let err = guarded_post_form(
&reqwest::Client::builder().build().unwrap(),
&format!("http://post-redirect.test:{}/token", addr.port()),
&[],
&[
("grant_type", "authorization_code"),
("code", "SECRET-CODE"),
],
)
.await
.expect_err("a POST followed a redirect");
let rendered = format!("{err:#}");
assert!(
rendered.contains("re-send the request body"),
"refused for the wrong reason: {rendered}"
);
assert_eq!(log.lock().unwrap().len(), 1);
assert_eq!(
elsewhere_log.lock().unwrap().len(),
0,
"the body was re-sent to the host the response chose"
);
}
#[tokio::test]
async fn too_many_redirects_is_refused() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let hops: Vec<String> = (0..MAX_REDIRECTS + 2)
.map(|i| redirect_to(&format!("http://redirect-loop.test:{port}/hop{i}")))
.collect();
let (addr, log) = spawn_http_on(listener, hops).await;
test_host_override("redirect-loop.test", addr);
let err = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://redirect-loop.test:{port}/hop0"),
&[],
)
.await
.expect_err("an endless redirect chain was not refused");
let rendered = format!("{err:#}");
assert!(
rendered.contains("too many redirects"),
"refused for the wrong reason: {rendered}"
);
assert_eq!(
log.lock().unwrap().len(),
MAX_REDIRECTS + 1,
"the guard made a different number of requests than its budget allows"
);
}
#[test]
fn the_content_type_follows_the_body_kind() {
assert_eq!(
PostBody::Json(b"{}".to_vec()).content_type(),
"application/json"
);
assert_eq!(
PostBody::Form(&[("a", "b")]).content_type(),
"application/x-www-form-urlencoded"
);
assert_eq!(
PostBody::Json(b"{\"a\":1}".to_vec()).into_bytes(),
b"{\"a\":1}"
);
assert_eq!(PostBody::Form(&[("a", "b c")]).into_bytes(), b"a=b+c");
}
#[test]
fn form_body_percent_encodes_and_cannot_inject_parameters() {
let body = PostBody::Form(&[
("grant_type", "authorization_code"),
("code", "abc&scope=evil"),
("redirect_uri", "https://x.example/oauth/callback"),
])
.into_bytes();
let s = String::from_utf8(body).unwrap();
assert!(s.contains("grant_type=authorization_code"));
assert!(
s.matches("scope=").count() == 0,
"a `&` in a value injected a parameter: {s}"
);
assert!(s.contains("%26"), "the `&` was not encoded: {s}");
assert!(s.contains("%3A%2F%2F"), "the `://` was not encoded: {s}");
}
async fn spawn_http(
responses: Vec<String>,
) -> (SocketAddr, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
spawn_http_on(listener, responses).await
}
async fn spawn_http_on(
listener: tokio::net::TcpListener,
responses: Vec<String>,
) -> (SocketAddr, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let addr = listener.local_addr().unwrap();
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let sink = std::sync::Arc::clone(&log);
tokio::spawn(async move {
let mut i = 0usize;
loop {
let Ok((mut sock, _)) = listener.accept().await else {
break;
};
let mut raw: Vec<u8> = Vec::new();
let mut chunk = [0u8; 4096];
loop {
let Ok(n) = sock.read(&mut chunk).await else {
return;
};
if n == 0 {
break;
}
raw.extend_from_slice(&chunk[..n]);
let Some(split) = raw.windows(4).position(|w| w == b"\r\n\r\n") else {
continue;
};
let (head, body) = raw.split_at(split + 4);
let want = String::from_utf8_lossy(head).lines().find_map(|l| {
let (k, v) = l.split_once(':')?;
k.eq_ignore_ascii_case("content-length")
.then(|| v.trim().parse::<usize>().ok())?
});
if want.is_none_or(|want| body.len() >= want) {
break;
}
}
if raw.is_empty() {
continue;
}
sink.lock()
.unwrap()
.push(String::from_utf8_lossy(&raw).to_string());
let body = responses
.get(i)
.cloned()
.unwrap_or_else(|| responses.last().cloned().unwrap_or_default());
i += 1;
let _ = sock.write_all(body.as_bytes()).await;
let _ = sock.flush().await;
}
});
(addr, log)
}
#[tokio::test]
async fn the_pinned_client_ignores_ambient_proxy_configuration() {
const VETTED: &str = "FR_AMBIENT_PROXY_VETTED";
const HOST: &str = "ambient-proxy-probe.invalid";
if let Ok(vetted) = std::env::var(VETTED) {
let addr: SocketAddr = vetted.parse().unwrap();
let client = build_pinned_client(HOST, addr).expect("client");
let _ = client.get(format!("http://{HOST}/probe")).send().await;
return;
}
let (vetted_addr, vetted_log) = spawn_http(vec![ok_200()]).await;
let (proxy_addr, proxy_log) = spawn_http(vec![ok_200()]).await;
let out = tokio::process::Command::new(std::env::current_exe().unwrap())
.args([
"net::tests::the_pinned_client_ignores_ambient_proxy_configuration",
"--exact",
"--test-threads=1",
])
.env_remove("NO_PROXY")
.env_remove("no_proxy")
.env_remove("REQUEST_METHOD")
.env(VETTED, vetted_addr.to_string())
.env("HTTP_PROXY", format!("http://{proxy_addr}"))
.env("HTTPS_PROXY", format!("http://{proxy_addr}"))
.env("ALL_PROXY", format!("http://{proxy_addr}"))
.output()
.await
.expect("re-exec the test binary");
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
out.status.success(),
"child run failed: {}",
String::from_utf8_lossy(&out.stderr)
);
assert!(
stdout.contains("1 passed"),
"the child ran no test, so this proves nothing about proxies — \
check the --exact filter. Child stdout:\n{stdout}"
);
let proxied = proxy_log.lock().unwrap().clone();
let direct = vetted_log.lock().unwrap().len();
assert!(
proxied.is_empty(),
"the pinned client used an ambient proxy, so the connect pin and \
`is_forbidden_ip` were both bypassed — the proxy resolves the \
hostname itself. Proxy saw: {proxied:?}"
);
assert_eq!(
direct, 1,
"the pinned client did not reach the vetted address it was pinned to",
);
}
fn ok_200() -> String {
"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nhi".to_string()
}
fn redirect_to(loc: &str) -> String {
format!("HTTP/1.1 302 Found\r\nLocation: {loc}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
}
fn not_modified_304() -> String {
"HTTP/1.1 304 Not Modified\r\nETag: \"v1\"\r\nConnection: close\r\n\r\n".to_string()
}
fn use_proxy_305(loc: &str) -> String {
format!("HTTP/1.1 305 Use Proxy\r\nLocation: {loc}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
}
#[tokio::test]
async fn an_unfollowable_3xx_is_refused_rather_than_returned() {
let (addr, log) = spawn_http(vec![use_proxy_305("http://proxy.invalid:3128/")]).await;
test_host_override("use-proxy.test", addr);
let err = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://use-proxy.test:{}/feed.xml", addr.port()),
&[],
)
.await
.expect_err("a 305 was returned to the caller instead of refused");
let msg = format!("{err:#}");
assert!(
msg.contains("305") && msg.contains("no single target"),
"refused, but not as an unfollowable status: {msg}",
);
assert_eq!(log.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn a_304_reaches_the_caller_instead_of_being_read_as_a_redirect() {
let (addr, log) = spawn_http(vec![not_modified_304()]).await;
test_host_override("not-modified.test", addr);
let resp = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://not-modified.test:{}/feed.xml", addr.port()),
&[],
)
.await
.expect("a 304 was treated as a redirect");
assert_eq!(
resp.status(),
reqwest::StatusCode::NOT_MODIFIED,
"the 304 did not survive the guard intact",
);
assert_eq!(
log.lock().unwrap().len(),
1,
"a 304 caused more than one request — it was followed, not returned",
);
}
#[tokio::test]
async fn a_302_is_still_followed_after_the_304_narrowing() {
let (b_addr, _b_log) = spawn_http(vec![ok_200()]).await;
test_host_override("still-follows-b.test", b_addr);
let (a_addr, a_log) = spawn_http(vec![redirect_to(&format!(
"http://still-follows-b.test:{}/final",
b_addr.port()
))])
.await;
test_host_override("still-follows-a.test", a_addr);
let resp = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://still-follows-a.test:{}/feed.xml", a_addr.port()),
&[],
)
.await
.expect("the 302 was not followed");
assert_eq!(resp.status(), reqwest::StatusCode::OK);
assert_eq!(a_log.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn a_redirect_to_a_forbidden_address_is_refused() {
let (addr, log) = spawn_http(vec![redirect_to(
"http://169.254.169.254/latest/meta-data/",
)])
.await;
test_host_override("hop-forbidden.test", addr);
let err = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://hop-forbidden.test:{}/feed.xml", addr.port()),
&[],
)
.await
.expect_err("a 302 to the metadata endpoint was followed");
let msg = format!("{err:#}");
assert!(
msg.contains("169.254.169.254") && msg.contains("forbidden"),
"refused, but not by the address check: {msg}",
);
assert_eq!(log.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn credentials_are_dropped_when_a_redirect_leaves_the_origin() {
let (b_addr, b_log) = spawn_http(vec![ok_200()]).await;
test_host_override("cred-b.test", b_addr);
let (a_addr, _a_log) = spawn_http(vec![redirect_to(&format!(
"http://cred-b.test:{}/next",
b_addr.port()
))])
.await;
test_host_override("cred-a.test", a_addr);
let resp = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://cred-a.test:{}/feed.xml", a_addr.port()),
&[(
HeaderName::from_static("authorization"),
HeaderValue::from_static("Bearer super-secret"),
)],
)
.await
.expect("the cross-origin hop should still succeed, just without the token");
assert!(resp.status().is_success());
let seen = b_log.lock().unwrap().join("\n").to_ascii_lowercase();
assert!(
seen.contains("get /next"),
"hop B recorded no request, so the assertions below prove nothing:\n{seen}",
);
assert!(
!seen.contains("super-secret"),
"the bearer token was forwarded across origins:\n{seen}",
);
assert!(
!seen.contains("authorization:"),
"the Authorization header survived a cross-origin redirect:\n{seen}",
);
}
#[tokio::test]
async fn credentials_survive_a_same_origin_redirect() {
let (addr, log) = spawn_http(vec![redirect_to("/second"), ok_200()]).await;
test_host_override("cred-same.test", addr);
let resp = guarded_get(
&reqwest::Client::builder().build().unwrap(),
&format!("http://cred-same.test:{}/feed.xml", addr.port()),
&[(
HeaderName::from_static("authorization"),
HeaderValue::from_static("Bearer keep-me"),
)],
)
.await
.expect("a same-origin redirect should be followed");
assert!(resp.status().is_success());
let reqs = log.lock().unwrap().clone();
assert_eq!(reqs.len(), 2, "the redirect was not followed");
assert!(
reqs[1].to_ascii_lowercase().contains("keep-me"),
"the token was stripped on a SAME-origin redirect — over-stripping \
would break every authenticated fetch:\n{}",
reqs[1],
);
}
#[test]
fn a_mixed_dns_answer_set_is_rejected_wholesale() {
let public: SocketAddr = "1.2.3.4:80".parse().unwrap();
let private: SocketAddr = "127.0.0.1:80".parse().unwrap();
let link_local: SocketAddr = "169.254.169.254:80".parse().unwrap();
assert_eq!(
first_vetted(
"ok.example",
[public, "5.6.7.8:80".parse().unwrap()].into_iter()
)
.unwrap(),
public,
);
for bad in [private, link_local] {
assert!(
first_vetted("evil.example", [public, bad].into_iter()).is_err(),
"{bad} in the answer set was accepted because a good answer came first",
);
assert!(first_vetted("evil.example", [bad, public].into_iter()).is_err());
}
assert!(first_vetted("empty.example", std::iter::empty()).is_err());
}
#[tokio::test]
async fn the_request_capture_is_not_truncated_at_the_buffer_size() {
let (addr, log) = spawn_http(vec![ok_200()]).await;
test_host_override("big-body.test", addr);
let filler = "x".repeat(32 * 1024);
let body = format!("{{\"pad\":\"{filler}\",\"tail\":\"THE-LAST-BYTES\"}}");
guarded_post_json(
&reqwest::Client::builder().build().unwrap(),
&format!("http://big-body.test:{}/ingest", addr.port()),
&[],
body.into_bytes(),
)
.await
.expect("the POST to the test server failed before anything was captured");
let seen = log.lock().unwrap().join("\n");
assert!(
seen.contains("POST /ingest"),
"the server captured no request at all: {} bytes",
seen.len()
);
assert!(
seen.contains("THE-LAST-BYTES"),
"the capture stops short of the request's end, so every negative \
assertion over it — including the one about not leaking an \
Authorization header across origins — can pass for the wrong \
reason. captured {} bytes",
seen.len()
);
}
const VERDICT_ATTEMPTS: usize = 3;
fn is_timeout(err: &anyhow::Error) -> bool {
err.downcast_ref::<reqwest::Error>()
.is_some_and(reqwest::Error::is_timeout)
}
async fn guarded_get_for_a_verdict(client: &Client, url: &str) -> Result<Response> {
for _ in 1..VERDICT_ATTEMPTS {
match guarded_get(client, url, &[]).await {
Err(err) if is_timeout(&err) => {
eprintln!("asking {url} again after a timeout, not a verdict: {err:#}");
}
verdict => return verdict,
}
}
guarded_get(client, url, &[]).await
}
#[tokio::test]
async fn a_timeout_is_recognised_as_a_timeout_and_not_a_certificate_verdict() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let mut held = Vec::new();
while let Ok((sock, _)) = listener.accept().await {
held.push(sock);
}
});
let client = Client::builder()
.no_proxy()
.timeout(Duration::from_millis(250))
.build()
.unwrap();
let raw = client
.get(format!("http://{addr}/never"))
.send()
.await
.expect_err("a server that never answers must not produce a response");
assert!(
raw.is_timeout(),
"the silent server ended the request some other way, so this test is \
not exercising a timeout at all: {raw}",
);
let err = anyhow::Error::from(raw).context(format!("fetching http://{addr}/never"));
assert!(
is_timeout(&err),
"a real read timeout was not recognised as one, so the retry would \
never retry and the certificate test stays latency-sensitive: {err:#}",
);
}
#[tokio::test]
async fn the_test_ca_is_trusted_and_still_validates_hostnames() {
let (addr, _log) = spawn_tls(|_| {
let mut r = std::collections::HashMap::new();
r.insert("/ok".to_string(), vec![TestResponse::json(200, "{}")]);
r
})
.await;
test_host_override("feed-tls.test", addr);
test_host_override("not-in-san.test", addr);
let client = reqwest::Client::builder().build().unwrap();
let ok = guarded_get_for_a_verdict(
&client,
&format!("https://feed-tls.test:{}/ok", addr.port()),
)
.await
.expect("a SAN-matching https host should be accepted");
assert!(ok.status().is_success());
let err = guarded_get_for_a_verdict(
&client,
&format!("https://not-in-san.test:{}/ok", addr.port()),
)
.await
.expect_err("a host with no SAN must still fail: validation is NOT disabled");
assert!(
!is_timeout(&err),
"a certificate verdict was classified as a timeout, so the retry would retry a genuine validation failure: {err:#}",
);
let msg = format!("{err:#}").to_ascii_lowercase();
assert!(
msg.contains("notvalidforname") || msg.contains("invalid peer certificate"),
"failed, but not because the certificate is invalid for this name — a \
DNS or connect failure would prove nothing here: {msg}",
);
}
}