use std::net::{IpAddr, SocketAddr};
use pingora_core::upstreams::peer::HttpPeer;
use super::peer::{AddressResolutionError, resolve_host_cached};
use crate::{connectivity::normalize_mapped_ipv4, subrequest::SubRequest};
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum UrlTargetError {
#[error(transparent)]
InvalidTarget(#[from] InvalidTarget),
#[error(transparent)]
Resolve(#[from] AddressResolutionError),
#[error("address policy rejected the resolved target")]
PolicyRejected(#[source] Box<dyn std::error::Error + Send + Sync>),
#[error("target preparation deadline exceeded")]
DeadlineExceeded,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum InvalidTarget {
#[error("malformed target: {0}")]
Malformed(String),
#[error("unsupported scheme: {0}")]
UnsupportedScheme(String),
#[error("target has no host")]
MissingHost,
#[error("target must not contain userinfo")]
UserinfoPresent,
#[error("target must not contain a fragment")]
FragmentPresent,
#[error("invalid port: {0}")]
InvalidPort(String),
#[error("invalid host literal: {0}")]
InvalidHost(String),
}
pub(crate) trait HostResolver {
fn resolve_host(&self, host: &str) -> impl Future<Output = Result<Vec<IpAddr>, AddressResolutionError>> + Send;
}
pub(crate) struct SystemResolver;
impl HostResolver for SystemResolver {
async fn resolve_host(&self, host: &str) -> Result<Vec<IpAddr>, AddressResolutionError> {
resolve_host_cached(host).await
}
}
pub(crate) struct ParsedTarget {
pub(crate) is_tls: bool,
pub(crate) host_authority: http::HeaderValue,
pub(crate) resolution_host: String,
pub(crate) effective_port: u16,
pub(crate) origin_form: http::Uri,
pub(crate) literal_ip: Option<IpAddr>,
}
impl std::fmt::Debug for ParsedTarget {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ParsedTarget")
.field("is_tls", &self.is_tls)
.field("host_authority", &self.host_authority)
.field("resolution_host", &self.resolution_host)
.field("effective_port", &self.effective_port)
.field("path", &self.origin_form.path())
.field("literal_ip", &self.literal_ip)
.finish()
}
}
#[expect(clippy::too_many_lines, reason = "comprehensive validation logic from spec §4.3")]
#[expect(
clippy::map_err_ignore,
reason = "error is intentionally discarded; we only care that parsing failed"
)]
pub(crate) fn parse_target(url: &str) -> Result<ParsedTarget, InvalidTarget> {
if url.contains('#') {
return Err(InvalidTarget::FragmentPresent);
}
let uri: http::Uri = url
.parse()
.map_err(|parse_err: http::uri::InvalidUri| InvalidTarget::Malformed(parse_err.to_string()))?;
let is_tls = match uri.scheme_str() {
Some("http") => false,
Some("https") => true,
Some(other) => return Err(InvalidTarget::UnsupportedScheme(other.to_owned())),
None => return Err(InvalidTarget::Malformed("missing scheme".to_owned())),
};
let authority = uri.authority().ok_or(InvalidTarget::MissingHost)?;
if authority.as_str().contains('@') {
return Err(InvalidTarget::UserinfoPresent);
}
let host = authority.host();
let (bracketed, bare) = match host.strip_prefix('[') {
Some(rest) => match rest.strip_suffix(']') {
Some(inner) => (true, inner),
None => return Err(InvalidTarget::Malformed("unterminated IPv6 literal".to_owned())),
},
None => (false, host),
};
if bare.is_empty() {
return Err(InvalidTarget::MissingHost);
}
let literal_ip = if bracketed {
let ip = bare
.parse::<IpAddr>()
.map_err(|_| InvalidTarget::InvalidHost(bare.to_owned()))?;
let IpAddr::V6(v6) = ip else {
return Err(InvalidTarget::InvalidHost(bare.to_owned()));
};
if v6.to_ipv4_mapped().is_some() {
return Err(InvalidTarget::InvalidHost(bare.to_owned()));
}
Some(IpAddr::V6(v6))
} else {
bare.parse::<IpAddr>().ok()
};
let effective_port = {
let remainder = authority.as_str().get(host.len()..).unwrap_or("");
match remainder.strip_prefix(':') {
None => {
if is_tls {
443
} else {
80
}
},
Some(token) => match token.parse::<u16>() {
Ok(port) if port >= 1 => port,
_ => return Err(InvalidTarget::InvalidPort(token.to_owned())),
},
}
};
let host_authority = http::HeaderValue::from_str(authority.as_str())
.map_err(|header_err| InvalidTarget::Malformed(header_err.to_string()))?;
let origin_form: http::Uri = match uri.query() {
Some(query) => format!("{}?{query}", uri.path()),
None => uri.path().to_owned(),
}
.parse()
.map_err(|parse_err: http::uri::InvalidUri| InvalidTarget::Malformed(parse_err.to_string()))?;
Ok(ParsedTarget {
is_tls,
host_authority,
resolution_host: bare.to_owned(),
effective_port,
origin_form,
literal_ip,
})
}
pub fn validate_url_target(url: &str) -> Result<(), InvalidTarget> {
parse_target(url).map(|_target| ())
}
pub struct PreparedTarget {
is_tls: bool,
host_authority: http::HeaderValue,
sni: String,
origin_form: http::Uri,
addresses: Vec<SocketAddr>,
}
impl std::fmt::Debug for PreparedTarget {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PreparedTarget")
.field("is_tls", &self.is_tls)
.field("host_authority", &self.host_authority)
.field("sni", &self.sni)
.field("path", &self.origin_form.path())
.field("addresses", &self.addresses)
.finish()
}
}
impl PreparedTarget {
pub(crate) fn new(
is_tls: bool,
host_authority: http::HeaderValue,
sni: String,
origin_form: http::Uri,
addresses: Vec<SocketAddr>,
) -> Self {
Self {
is_tls,
host_authority,
sni: if is_tls { sni } else { String::new() },
origin_form,
addresses,
}
}
#[must_use]
pub fn bind(self, mut request: SubRequest) -> PreparedSubrequest {
request.headers.insert(http::header::HOST, self.host_authority);
request.uri = self.origin_form;
PreparedSubrequest {
is_tls: self.is_tls,
sni: self.sni,
addresses: self.addresses,
request,
}
}
#[must_use]
pub fn addresses(&self) -> &[SocketAddr] {
&self.addresses
}
#[must_use]
pub fn host_authority(&self) -> &http::HeaderValue {
&self.host_authority
}
#[must_use]
pub fn is_tls(&self) -> bool {
self.is_tls
}
#[must_use]
pub fn sni(&self) -> &str {
&self.sni
}
}
pub struct PreparedSubrequest {
is_tls: bool,
sni: String,
addresses: Vec<SocketAddr>,
request: SubRequest,
}
impl PreparedSubrequest {
pub fn peers(&self) -> impl Iterator<Item = HttpPeer> + '_ {
self.addresses
.iter()
.map(|addr| peer_for(*addr, self.is_tls, &self.sni))
}
#[must_use]
pub fn peer_at(&self, index: usize) -> Option<HttpPeer> {
self.addresses
.get(index)
.map(|addr| peer_for(*addr, self.is_tls, &self.sni))
}
#[must_use]
pub fn addresses(&self) -> &[SocketAddr] {
&self.addresses
}
#[must_use]
pub fn request(&self) -> &SubRequest {
&self.request
}
}
fn peer_for(addr: SocketAddr, is_tls: bool, sni: &str) -> HttpPeer {
if is_tls {
assert!(
!sni.is_empty(),
"BUG: refusing to build a TLS peer with an empty SNI (non-empty SNI invariant violated)"
);
HttpPeer::new(addr, true, sni.to_owned())
} else {
HttpPeer::new(addr, false, String::new())
}
}
pub async fn prepare_url_target<F>(
url: &str,
deadline: std::time::Instant,
validate: F,
) -> Result<PreparedTarget, UrlTargetError>
where
F: FnOnce(&[SocketAddr]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
{
prepare_url_target_with_resolver(url, deadline, validate, &SystemResolver).await
}
#[expect(
clippy::too_many_lines,
reason = "linear pin-before-dial pipeline with four deadline checkpoints; splitting would break the one-clock invariant"
)]
pub(crate) async fn prepare_url_target_with_resolver<R: HostResolver + Sync>(
url: &str,
deadline: std::time::Instant,
validate: impl FnOnce(&[SocketAddr]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
resolver: &R,
) -> Result<PreparedTarget, UrlTargetError> {
let deadline = tokio::time::Instant::from_std(deadline);
if tokio::time::Instant::now() >= deadline {
return Err(UrlTargetError::DeadlineExceeded);
}
let parsed = parse_target(url)?;
if tokio::time::Instant::now() >= deadline {
return Err(UrlTargetError::DeadlineExceeded);
}
let raw_ips: Vec<IpAddr> = match parsed.literal_ip {
Some(ip) => vec![ip],
None => tokio::time::timeout_at(deadline, resolver.resolve_host(&parsed.resolution_host))
.await
.map_err(|_elapsed| UrlTargetError::DeadlineExceeded)??,
};
let mut seen = std::collections::HashSet::new();
let addresses: Vec<SocketAddr> = raw_ips
.into_iter()
.map(normalize_mapped_ipv4)
.map(|ip| SocketAddr::new(ip, parsed.effective_port))
.filter(|sa| seen.insert(*sa))
.collect();
if addresses.is_empty() {
return Err(UrlTargetError::Resolve(AddressResolutionError::Empty(
parsed.resolution_host,
)));
}
if tokio::time::Instant::now() >= deadline {
return Err(UrlTargetError::DeadlineExceeded);
}
validate(&addresses).map_err(UrlTargetError::PolicyRejected)?;
if tokio::time::Instant::now() >= deadline {
return Err(UrlTargetError::DeadlineExceeded);
}
Ok(PreparedTarget::new(
parsed.is_tls,
parsed.host_authority,
parsed.resolution_host,
parsed.origin_form,
addresses,
))
}
#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::indexing_slicing,
clippy::assertions_on_result_states,
clippy::min_ident_chars,
clippy::shadow_unrelated,
clippy::arithmetic_side_effects,
reason = "tests"
)]
mod tests {
use std::time::{Duration, Instant};
use super::*;
use crate::subrequest::{StreamLimits, SubRequestClient, SubRequestConnector};
#[test]
fn invalid_target_display_is_token_only() {
assert_eq!(
InvalidTarget::UnsupportedScheme("ftp".to_owned()).to_string(),
"unsupported scheme: ftp"
);
assert_eq!(
InvalidTarget::InvalidPort("99999".to_owned()).to_string(),
"invalid port: 99999"
);
assert!(!InvalidTarget::UserinfoPresent.to_string().contains('@'));
}
#[test]
fn policy_rejected_display_hides_the_source() {
let source: Box<dyn std::error::Error + Send + Sync> = "secret-token-abc".into();
let err = UrlTargetError::PolicyRejected(source);
assert!(
!err.to_string().contains("secret-token-abc"),
"PolicyRejected Display must not format caller-supplied data: {err}"
);
}
#[test]
fn invalid_target_converts_into_url_target_error() {
let err: UrlTargetError = InvalidTarget::MissingHost.into();
assert!(matches!(err, UrlTargetError::InvalidTarget(InvalidTarget::MissingHost)));
}
#[expect(clippy::panic, reason = "test utility panics to fail the test on parse error")]
fn parse_ok(url: &str) -> ParsedTarget {
parse_target(url).unwrap_or_else(|e| panic!("expected {url} to parse: {e}"))
}
#[test]
fn parses_https_default_port_and_host_header() {
let p = parse_ok("https://api.example.com/v1/models?a=1");
assert!(p.is_tls);
assert_eq!(p.effective_port, 443);
assert_eq!(p.host_authority.to_str().unwrap(), "api.example.com");
assert_eq!(p.resolution_host, "api.example.com");
assert_eq!(p.origin_form.to_string(), "/v1/models?a=1");
assert!(p.literal_ip.is_none());
}
#[test]
fn parses_http_default_port() {
let p = parse_ok("http://example.com/");
assert!(!p.is_tls);
assert_eq!(p.effective_port, 80);
assert_eq!(p.origin_form.to_string(), "/");
}
#[test]
fn explicit_non_default_port_flows_to_host_and_dial() {
let p = parse_ok("https://example.com:8443/x");
assert_eq!(p.effective_port, 8443);
assert_eq!(p.host_authority.to_str().unwrap(), "example.com:8443");
assert_eq!(p.resolution_host, "example.com");
}
#[test]
fn parses_ipv4_literal_as_literal_ip() {
let p = parse_ok("http://127.0.0.1/");
assert_eq!(p.literal_ip, Some("127.0.0.1".parse::<IpAddr>().unwrap()));
assert_eq!(p.effective_port, 80);
assert_eq!(p.resolution_host, "127.0.0.1");
assert_eq!(p.host_authority.to_str().unwrap(), "127.0.0.1");
}
#[test]
fn parses_bracketed_ipv6_literal() {
let p = parse_ok("https://[2001:db8::1]:8443/");
assert_eq!(p.literal_ip, Some("2001:db8::1".parse::<IpAddr>().unwrap()));
assert_eq!(p.resolution_host, "2001:db8::1");
assert_eq!(p.host_authority.to_str().unwrap(), "[2001:db8::1]:8443");
assert_eq!(p.effective_port, 8443);
}
#[test]
fn empty_path_defaults_to_slash() {
assert_eq!(parse_ok("http://h").origin_form.to_string(), "/");
}
#[test]
fn query_only_url_gets_root_path() {
let p = parse_ok("https://api.example.com?token=x");
assert_eq!(
p.origin_form.to_string(),
"/?token=x",
"an empty path with a query must default to /, not the leading-slash-less ?token=x"
);
}
#[test]
fn rejects_unsupported_scheme() {
assert!(matches!(
parse_target("ftp://h/"),
Err(InvalidTarget::UnsupportedScheme(s)) if s == "ftp"
));
}
#[test]
fn rejects_userinfo_including_empty() {
assert!(matches!(
parse_target("http://user:pw@h/"),
Err(InvalidTarget::UserinfoPresent)
));
assert!(matches!(
parse_target("http://@h/"),
Err(InvalidTarget::UserinfoPresent)
));
}
#[test]
fn rejects_fragment() {
assert!(matches!(
parse_target("http://h/p#frag"),
Err(InvalidTarget::FragmentPresent)
));
}
#[test]
fn rejects_empty_and_bracket_only_hosts() {
assert!(matches!(parse_target("https://[]/"), Err(InvalidTarget::MissingHost)));
assert!(matches!(
parse_target("https://[]:443/"),
Err(InvalidTarget::MissingHost)
));
assert!(matches!(parse_target("https://:443/"), Err(InvalidTarget::MissingHost)));
}
#[test]
#[expect(clippy::panic, reason = "test panics on unexpected parse result")]
fn rejects_malformed_ip_literals_with_token_only() {
for (url, token) in [("https://[gggg::1]/", "gggg::1"), ("https://[vFF.abc]/", "vFF.abc")] {
match parse_target(url) {
Err(InvalidTarget::InvalidHost(t)) => {
assert_eq!(t, token);
assert!(!t.contains("https://"), "InvalidHost must carry the token only");
},
other => panic!("expected InvalidHost({token}), got {other:?}"),
}
}
}
#[test]
fn rejects_ipv4_mapped_ipv6_literal() {
assert!(matches!(
parse_target("https://[::ffff:1.2.3.4]/"),
Err(InvalidTarget::InvalidHost(_))
));
}
#[test]
fn rejects_bracketed_ipv4_literal() {
assert!(
matches!(
parse_target("https://[127.0.0.1]/"),
Err(InvalidTarget::InvalidHost(t)) if t == "127.0.0.1"
),
"brackets denote an IPv6 literal (RFC 3986); a bracketed IPv4 is the malformed Host [127.0.0.1]"
);
}
#[test]
#[expect(clippy::panic, reason = "test panics on unexpected parse result")]
fn rejects_bad_ports_with_token_only() {
for (url, token) in [
("http://h:99999/", "99999"),
("http://h:abc/", "abc"),
("http://h:-1/", "-1"),
("http://h:0/", "0"),
("http://h:8080evil/", "8080evil"),
("http://[::1]:abc/", "abc"),
("http://[::1]:/", ""),
] {
match parse_target(url) {
Err(InvalidTarget::InvalidPort(t)) => assert_eq!(t, token, "for {url}"),
other => panic!("expected InvalidPort({token:?}) for {url}, got {other:?}"),
}
}
}
#[test]
fn parse_errors_never_contain_the_input_url() {
let url = "ftp://secret-user@host/path?token=abc";
let err = parse_target(url).unwrap_err();
assert!(!err.to_string().contains("secret-user"));
assert!(!err.to_string().contains("token=abc"));
let malformed = "https://exa mple.com/secret-path?q=leak";
let merr = parse_target(malformed).unwrap_err();
assert!(
matches!(merr, InvalidTarget::Malformed(_)),
"expected Malformed, got {merr:?}"
);
let disp = merr.to_string();
assert!(!disp.contains("exa mple"), "Malformed Display leaked host: {disp}");
assert!(!disp.contains("secret-path"), "Malformed Display leaked path: {disp}");
assert!(!disp.contains("q=leak"), "Malformed Display leaked query: {disp}");
}
use pingora_core::upstreams::peer::{Peer as _, Scheme};
fn target(is_tls: bool, authority: &str, sni: &str, addrs: &[&str]) -> PreparedTarget {
PreparedTarget::new(
is_tls,
http::HeaderValue::from_str(authority).unwrap(),
sni.to_owned(),
"/p?q=1".parse().unwrap(),
addrs.iter().map(|a| a.parse::<SocketAddr>().unwrap()).collect(),
)
}
fn req_with_host(host: &str) -> SubRequest {
let mut headers = http::HeaderMap::new();
headers.insert(http::header::HOST, http::HeaderValue::from_str(host).unwrap());
SubRequest {
method: http::Method::GET,
uri: "/original".parse().unwrap(),
headers,
body: bytes::Bytes::new(),
}
}
#[test]
fn accessors_expose_frozen_fields() {
let t = target(true, "api.example.com:8443", "api.example.com", &["93.184.216.34:8443"]);
assert!(t.is_tls());
assert_eq!(t.sni(), "api.example.com");
assert_eq!(t.host_authority().to_str().unwrap(), "api.example.com:8443");
assert_eq!(t.addresses(), ["93.184.216.34:8443".parse::<SocketAddr>().unwrap()]);
}
#[test]
fn http_target_has_empty_sni() {
let t = target(false, "example.com", "example.com", &["93.184.216.34:80"]);
assert_eq!(t.sni(), "", "sni() must be empty for http");
}
#[test]
fn bind_overwrites_host_and_rewrites_target() {
let t = target(true, "api.example.com:8443", "api.example.com", &["93.184.216.34:8443"]);
let prepared = t.bind(req_with_host("attacker.example"));
let req = prepared.request();
assert_eq!(
req.headers.get(http::header::HOST).unwrap().to_str().unwrap(),
"api.example.com:8443",
"bind must overwrite a conflicting caller Host"
);
assert_eq!(req.headers.get_all(http::header::HOST).iter().count(), 1);
assert_eq!(
req.uri.to_string(),
"/p?q=1",
"bind must rewrite the target to origin-form"
);
}
#[test]
fn peers_build_tls_peer_per_address_in_order() {
let t = target(true, "h:8443", "h.example.com", &["10.0.0.1:8443", "10.0.0.2:8443"]);
let prepared = t.bind(req_with_host("ignored"));
let peers: Vec<_> = prepared.peers().collect();
assert_eq!(peers.len(), 2);
assert_eq!(peers[0].address().to_string(), "10.0.0.1:8443");
assert_eq!(peers[1].address().to_string(), "10.0.0.2:8443");
assert_eq!(peers[0].sni, "h.example.com");
assert_eq!(peers[0].scheme, Scheme::HTTPS);
assert_eq!(prepared.peer_at(1).unwrap().address().to_string(), "10.0.0.2:8443");
assert!(prepared.peer_at(2).is_none());
}
#[test]
fn http_peers_have_empty_sni_and_plain_scheme() {
let t = target(false, "h", "", &["10.0.0.1:80"]);
let prepared = t.bind(req_with_host("ignored"));
let peer = prepared.peer_at(0).unwrap();
assert_eq!(peer.sni, "");
assert_eq!(peer.scheme, Scheme::HTTP);
}
#[test]
#[should_panic(expected = "non-empty SNI")]
fn peer_for_refuses_empty_tls_sni() {
drop(peer_for("10.0.0.1:443".parse().unwrap(), true, ""));
}
use std::sync::{
Arc as StdArc,
atomic::{AtomicUsize, Ordering},
};
#[derive(Clone)]
enum FakeDelay {
None,
Sleep(Duration),
Advance(Duration),
}
struct FakeResolver {
calls: StdArc<AtomicUsize>,
respond: Box<dyn Fn() -> Result<Vec<IpAddr>, AddressResolutionError> + Send + Sync>,
delay: FakeDelay,
}
impl FakeResolver {
fn ok(ips: Vec<IpAddr>) -> Self {
Self {
calls: StdArc::new(AtomicUsize::new(0)),
respond: Box::new(move || Ok(ips.clone())),
delay: FakeDelay::None,
}
}
fn failing() -> Self {
Self {
calls: StdArc::new(AtomicUsize::new(0)),
respond: Box::new(|| Err(AddressResolutionError::Empty("fake".to_owned()))),
delay: FakeDelay::None,
}
}
fn with_delay(mut self, delay: FakeDelay) -> Self {
self.delay = delay;
self
}
fn call_count(&self) -> usize {
self.calls.load(Ordering::SeqCst)
}
}
impl HostResolver for FakeResolver {
async fn resolve_host(&self, _host: &str) -> Result<Vec<IpAddr>, AddressResolutionError> {
self.calls.fetch_add(1, Ordering::SeqCst);
let out = (self.respond)();
let delay = self.delay.clone();
match delay {
FakeDelay::None => {},
FakeDelay::Sleep(d) => tokio::time::sleep(d).await,
FakeDelay::Advance(d) => tokio::time::advance(d).await,
}
out
}
}
#[tokio::test]
async fn system_resolver_resolves_localhost() {
let ips = SystemResolver
.resolve_host("localhost")
.await
.expect("localhost resolves");
assert!(ips.iter().all(IpAddr::is_loopback), "got {ips:?}");
}
#[tokio::test]
async fn fake_resolver_counts_calls() {
let fake = FakeResolver::ok(vec!["1.2.3.4".parse().unwrap()]);
let ips = fake.resolve_host("h").await.unwrap();
assert_eq!(ips, vec!["1.2.3.4".parse::<IpAddr>().unwrap()]);
assert_eq!(fake.call_count(), 1);
}
fn far_deadline() -> Instant {
Instant::now() + Duration::from_secs(30)
}
#[tokio::test]
async fn ip_literal_skips_dns_but_hits_hook() {
for url in ["http://127.0.0.1/", "http://[::1]/"] {
let fake = FakeResolver::ok(vec!["9.9.9.9".parse().unwrap()]);
let seen: std::sync::Mutex<Vec<SocketAddr>> = std::sync::Mutex::new(Vec::new());
let target = prepare_url_target_with_resolver(
url,
far_deadline(),
|addrs| {
seen.lock().unwrap().extend_from_slice(addrs);
Ok(())
},
&fake,
)
.await
.expect("literal resolves without DNS");
assert_eq!(fake.call_count(), 0, "IP literals must not touch the resolver");
assert_eq!(
seen.lock().unwrap().len(),
1,
"the hook must see the literal address for {url}"
);
assert_eq!(target.addresses().len(), 1);
}
}
#[tokio::test]
async fn rejecting_hook_yields_policy_rejected_and_no_target() {
let fake = FakeResolver::ok(vec![]);
let err = prepare_url_target_with_resolver(
"http://127.0.0.1/",
far_deadline(),
|_| Err("blocked by policy".into()),
&fake,
)
.await
.expect_err("a rejecting hook blocks the target");
assert!(matches!(err, UrlTargetError::PolicyRejected(_)), "got {err}");
}
#[tokio::test]
async fn dns_host_resolving_to_empty_set_yields_resolve_empty() {
let fake = FakeResolver::ok(vec![]);
let err = prepare_url_target_with_resolver("http://empty-set.test/", far_deadline(), |_| Ok(()), &fake)
.await
.expect_err("empty resolved set");
assert!(
matches!(err, UrlTargetError::Resolve(AddressResolutionError::Empty(_))),
"a zero-address DNS answer must be a clean Resolve(Empty), not a peerless target: {err:?}"
);
assert_eq!(fake.call_count(), 1, "the DNS host must hit the resolver exactly once");
}
#[tokio::test]
async fn complete_set_reaches_hook_normalized_and_deduped() {
let fake = FakeResolver::ok(vec![
"::ffff:1.2.3.4".parse().unwrap(),
"1.2.3.4".parse().unwrap(),
"9.9.9.9".parse().unwrap(),
]);
let target = prepare_url_target_with_resolver("https://h.example.com/", far_deadline(), |_| Ok(()), &fake)
.await
.expect("resolves");
assert_eq!(
target.addresses(),
[
"1.2.3.4:443".parse::<SocketAddr>().unwrap(),
"9.9.9.9:443".parse::<SocketAddr>().unwrap(),
],
"mapped-IPv4 unwrapped, order-preserving dedup, effective port applied"
);
}
#[tokio::test]
async fn resolver_error_becomes_url_resolve_error_without_calling_hook() {
let fake = FakeResolver::failing();
let called = StdArc::new(std::sync::atomic::AtomicBool::new(false));
let called_hook = StdArc::clone(&called);
let err = prepare_url_target_with_resolver(
"https://h.example.com/",
far_deadline(),
move |_| {
called_hook.store(true, Ordering::SeqCst);
Ok(())
},
&fake,
)
.await
.expect_err("resolver failure surfaces");
assert!(matches!(err, UrlTargetError::Resolve(_)), "got {err}");
assert!(
!called.load(Ordering::SeqCst),
"the hook must not run on resolution failure"
);
}
#[tokio::test]
async fn no_re_resolve_across_fallback() {
let fake = FakeResolver::ok(vec!["1.2.3.4".parse().unwrap(), "5.6.7.8".parse().unwrap()]);
let target = prepare_url_target_with_resolver("http://h.example.com/", far_deadline(), |_| Ok(()), &fake)
.await
.expect("resolves");
let prepared = target.bind(req_with_host("ignored"));
let _peers: Vec<_> = prepared.peers().collect();
assert_eq!(fake.call_count(), 1, "fallback must not re-resolve");
}
#[tokio::test]
async fn deadline_expired_on_entry_for_literal() {
let past = Instant::now();
tokio::time::sleep(Duration::from_millis(10)).await;
let fake = FakeResolver::ok(vec![]);
let err = prepare_url_target_with_resolver("http://127.0.0.1/", past, |_| Ok(()), &fake)
.await
.expect_err("elapsed deadline");
assert!(matches!(err, UrlTargetError::DeadlineExceeded));
}
#[tokio::test]
async fn deadline_expires_during_resolution() {
let fake =
FakeResolver::ok(vec!["1.2.3.4".parse().unwrap()]).with_delay(FakeDelay::Sleep(Duration::from_secs(60)));
let deadline = Instant::now() + Duration::from_millis(50);
let err = prepare_url_target_with_resolver("https://h.example.com/", deadline, |_| Ok(()), &fake)
.await
.expect_err("resolution outruns the deadline");
assert!(
matches!(err, UrlTargetError::DeadlineExceeded),
"timeout_at must wrap resolve_host: {err}"
);
}
#[tokio::test(start_paused = true)]
async fn deadline_checkpoint_before_validate_fires() {
let deadline = Instant::now() + Duration::from_secs(30);
let fake =
FakeResolver::ok(vec!["1.2.3.4".parse().unwrap()]).with_delay(FakeDelay::Advance(Duration::from_secs(31)));
let called = StdArc::new(std::sync::atomic::AtomicBool::new(false));
let called_hook = StdArc::clone(&called);
let err = prepare_url_target_with_resolver(
"https://h.example.com/",
deadline,
move |_| {
called_hook.store(true, Ordering::SeqCst);
Ok(())
},
&fake,
)
.await
.expect_err("checkpoint 3 catches expiry");
assert!(matches!(err, UrlTargetError::DeadlineExceeded));
assert!(
!called.load(Ordering::SeqCst),
"pre-validate checkpoint must block the hook"
);
}
#[tokio::test]
#[allow(
clippy::disallowed_methods,
reason = "synchronous hook intentionally blocks the real clock to test checkpoint 4"
)]
async fn deadline_after_validate_returns_deadline_but_hook_ran() {
let deadline = Instant::now() + Duration::from_millis(30);
let fake = FakeResolver::ok(vec![]);
let ran = StdArc::new(std::sync::atomic::AtomicBool::new(false));
let ran_hook = StdArc::clone(&ran);
let err = prepare_url_target_with_resolver(
"http://127.0.0.1/",
deadline,
move |_| {
ran_hook.store(true, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(80));
Ok(())
},
&fake,
)
.await
.expect_err("checkpoint 4 catches post-hook expiry");
assert!(matches!(err, UrlTargetError::DeadlineExceeded));
assert!(
ran.load(Ordering::SeqCst),
"the synchronous hook still ran to completion"
);
}
#[tokio::test]
async fn public_wrapper_prepares_a_literal_target() {
let target = prepare_url_target("http://127.0.0.1:9/health", far_deadline(), |_| Ok(()))
.await
.expect("literal prepares");
assert!(!target.is_tls());
assert_eq!(target.addresses(), ["127.0.0.1:9".parse::<SocketAddr>().unwrap()]);
assert_eq!(target.host_authority().to_str().unwrap(), "127.0.0.1:9");
}
async fn spawn_loopback_backend(body: &'static str) -> u16 {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
tokio::spawn(async move {
loop {
let Ok((mut stream, _)) = listener.accept().await else {
return;
};
let response = body.to_owned();
tokio::spawn(async move {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let mut buf = [0_u8; 1024];
drop(stream.read(&mut buf).await);
let msg = format!(
"HTTP/1.1 200 OK\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
response.len(),
response
);
drop(stream.write_all(msg.as_bytes()).await);
drop(stream.flush().await);
});
}
});
port
}
fn get_request() -> SubRequest {
SubRequest {
method: http::Method::GET,
uri: "/".parse().unwrap(),
headers: http::HeaderMap::new(),
body: bytes::Bytes::new(),
}
}
#[tokio::test]
#[expect(
clippy::too_many_lines,
reason = "buffered + streaming parity in one test proves both share the prepared target"
)]
async fn buffered_and_streaming_share_the_prepared_target() {
let port = spawn_loopback_backend("hello-parity").await;
let url = format!("http://127.0.0.1:{port}/");
let deadline = Instant::now() + Duration::from_secs(5);
praxis_tls::provider::install();
let client = SubRequestClient::new(SubRequestConnector::new(1, None));
let target = prepare_url_target(&url, deadline, |_| Ok(())).await.unwrap();
let prepared = target.bind(get_request());
let peer = prepared.peer_at(0).expect("one address");
let resp = Box::pin(client.execute(&peer, prepared.request(), 1_048_576, Duration::from_secs(5), None))
.await
.expect("buffered exchange");
assert_eq!(resp.status, 200);
assert_eq!(resp.body, bytes::Bytes::from_static(b"hello-parity"));
let target = prepare_url_target(&url, deadline, |_| Ok(())).await.unwrap();
let prepared = target.bind(get_request());
let peer = prepared.peer_at(0).expect("one address");
let limits = StreamLimits {
idle_timeout: Duration::from_secs(5),
max_stream_duration: None,
max_total_bytes: None,
};
let mut streaming =
Box::pin(client.send_streaming(&peer, prepared.request(), Duration::from_secs(5), limits, None))
.await
.expect("streaming exchange");
assert_eq!(streaming.status, 200);
let mut collected = Vec::new();
while let Some(chunk) = streaming.body.next_chunk().await.expect("chunk") {
collected.extend_from_slice(&chunk);
}
assert_eq!(collected, b"hello-parity");
drop(streaming);
}
#[test]
fn url_target_error_debug() {
let err = UrlTargetError::DeadlineExceeded;
let debug = format!("{err:?}");
assert!(debug.contains("DeadlineExceeded"));
let invalid = UrlTargetError::InvalidTarget(InvalidTarget::MissingHost);
let debug = format!("{invalid:?}");
assert!(debug.contains("InvalidTarget"));
assert!(debug.contains("MissingHost"));
let resolve = UrlTargetError::Resolve(AddressResolutionError::Empty("test".to_owned()));
let debug = format!("{resolve:?}");
assert!(debug.contains("Resolve"));
assert!(debug.contains("Empty"));
}
#[test]
fn url_target_error_display() {
assert_eq!(
UrlTargetError::DeadlineExceeded.to_string(),
"target preparation deadline exceeded"
);
let err = UrlTargetError::InvalidTarget(InvalidTarget::MissingHost);
assert_eq!(err.to_string(), "target has no host");
let policy = UrlTargetError::PolicyRejected("test".into());
assert_eq!(policy.to_string(), "address policy rejected the resolved target");
}
#[test]
fn invalid_target_debug() {
let err = InvalidTarget::MissingHost;
let debug = format!("{err:?}");
assert!(debug.contains("MissingHost"));
let malformed = InvalidTarget::Malformed("test error".to_owned());
let debug = format!("{malformed:?}");
assert!(debug.contains("Malformed"));
assert!(debug.contains("test error"));
let scheme = InvalidTarget::UnsupportedScheme("ws".to_owned());
let debug = format!("{scheme:?}");
assert!(debug.contains("UnsupportedScheme"));
assert!(debug.contains("ws"));
}
#[test]
fn parsed_target_debug_redacts_query() {
let p = parse_ok("https://api.example.com/v1/models?token=secret123&key=value");
let debug = format!("{p:?}");
assert!(debug.contains("ParsedTarget"));
assert!(debug.contains("is_tls"));
assert!(debug.contains("true"));
assert!(debug.contains("api.example.com"));
assert!(debug.contains("/v1/models"), "path should be visible");
assert!(
!debug.contains("token=secret123"),
"query should be redacted from Debug"
);
assert!(!debug.contains("key=value"), "query should be redacted from Debug");
}
#[test]
fn prepared_target_debug_redacts_query() {
let t = target(true, "api.example.com:443", "api.example.com", &["93.184.216.34:443"]);
let debug = format!("{t:?}");
assert!(debug.contains("PreparedTarget"));
assert!(debug.contains("is_tls"));
assert!(debug.contains("true"));
assert!(debug.contains("api.example.com"));
assert!(debug.contains("/p"), "path should be visible");
assert!(!debug.contains("q=1"), "query should be redacted from Debug");
}
#[test]
fn validate_url_target_success() {
assert!(validate_url_target("http://example.com/").is_ok());
assert!(validate_url_target("https://api.example.com:8443/path").is_ok());
assert!(validate_url_target("http://127.0.0.1:9000/").is_ok());
assert!(validate_url_target("https://[::1]/").is_ok());
}
#[test]
fn validate_url_target_failures() {
assert!(validate_url_target("ftp://example.com/").is_err());
assert!(validate_url_target("http://user@host/").is_err());
assert!(validate_url_target("http://host/#fragment").is_err());
assert!(validate_url_target("http://:80/").is_err());
assert!(validate_url_target("http://host:99999/").is_err());
}
#[tokio::test]
async fn deadline_checkpoint_after_parse() {
tokio::time::pause();
let deadline = Instant::now() + Duration::from_millis(10);
tokio::time::advance(Duration::from_millis(11)).await;
let fake = FakeResolver::ok(vec![]);
let err = prepare_url_target_with_resolver("http://127.0.0.1/", deadline, |_| Ok(()), &fake)
.await
.expect_err("checkpoint 2 should fire");
assert!(matches!(err, UrlTargetError::DeadlineExceeded));
assert_eq!(fake.call_count(), 0, "checkpoint 2 happens before DNS");
}
#[tokio::test]
async fn prepared_subrequest_addresses_accessor() {
let t = target(false, "h", "", &["10.0.0.1:80", "10.0.0.2:80"]);
let prepared = t.bind(req_with_host("ignored"));
let addrs = prepared.addresses();
assert_eq!(addrs.len(), 2);
assert_eq!(addrs[0].to_string(), "10.0.0.1:80");
assert_eq!(addrs[1].to_string(), "10.0.0.2:80");
}
#[tokio::test]
async fn prepared_subrequest_request_accessor() {
let t = target(true, "api.example.com:443", "api.example.com", &["93.184.216.34:443"]);
let req = req_with_host("original.example");
let prepared = t.bind(req);
let bound_req = prepared.request();
assert_eq!(
bound_req.headers.get(http::header::HOST).unwrap().to_str().unwrap(),
"api.example.com:443",
"bound request should have URL authority as Host"
);
}
#[tokio::test]
async fn system_resolver_error_path() {
let result = SystemResolver
.resolve_host("nonexistent-host-12345-abcde.invalid")
.await;
assert!(result.is_err(), "nonexistent domain should fail to resolve");
}
#[test]
fn parses_port_65535() {
let p = parse_ok("http://example.com:65535/");
assert_eq!(p.effective_port, 65535);
}
#[test]
fn parses_ipv6_with_zone_id_rejected() {
let err = parse_target("http://[fe80::1%eth0]/").unwrap_err();
assert!(
matches!(err, InvalidTarget::InvalidHost(_)),
"zone IDs should be rejected: {err:?}"
);
}
#[test]
fn rejects_missing_scheme() {
let err = parse_target("example.com/").unwrap_err();
assert!(
matches!(err, InvalidTarget::Malformed(_)),
"missing scheme should be Malformed"
);
}
#[test]
fn rejects_unterminated_bracket() {
let err = parse_target("http://[::1/").unwrap_err();
assert!(
matches!(err, InvalidTarget::Malformed(_)),
"unterminated bracket should be Malformed: {err:?}"
);
}
#[test]
fn address_resolution_error_converts() {
let resolve_err = AddressResolutionError::Empty("test".to_owned());
let url_err: UrlTargetError = resolve_err.into();
assert!(matches!(url_err, UrlTargetError::Resolve(_)));
}
#[test]
fn prepared_target_new_with_http_clears_sni() {
let t = PreparedTarget::new(
false,
http::HeaderValue::from_str("example.com").unwrap(),
"example.com".to_owned(),
"/".parse().unwrap(),
vec!["93.184.216.34:80".parse().unwrap()],
);
assert_eq!(t.sni(), "", "http target should have empty SNI even if passed");
}
#[test]
fn prepared_target_new_with_https_keeps_sni() {
let t = PreparedTarget::new(
true,
http::HeaderValue::from_str("example.com:443").unwrap(),
"example.com".to_owned(),
"/".parse().unwrap(),
vec!["93.184.216.34:443".parse().unwrap()],
);
assert_eq!(t.sni(), "example.com");
}
#[tokio::test]
async fn multiple_addresses_all_reachable_via_peer_at() {
let t = target(
true,
"h:443",
"h.example.com",
&["10.0.0.1:443", "10.0.0.2:443", "10.0.0.3:443"],
);
let prepared = t.bind(req_with_host("ignored"));
assert_eq!(prepared.peer_at(0).unwrap().address().to_string(), "10.0.0.1:443");
assert_eq!(prepared.peer_at(1).unwrap().address().to_string(), "10.0.0.2:443");
assert_eq!(prepared.peer_at(2).unwrap().address().to_string(), "10.0.0.3:443");
assert!(prepared.peer_at(3).is_none());
assert!(prepared.peer_at(999).is_none());
}
#[test]
fn parses_query_with_multiple_params() {
let p = parse_ok("https://api.example.com/endpoint?a=1&b=2&c=3");
assert_eq!(p.origin_form.to_string(), "/endpoint?a=1&b=2&c=3");
}
#[test]
fn parses_encoded_characters_in_path() {
let p = parse_ok("https://example.com/path%20with%20spaces");
assert_eq!(p.origin_form.to_string(), "/path%20with%20spaces");
}
#[tokio::test]
async fn policy_rejected_preserves_source() {
let source: Box<dyn std::error::Error + Send + Sync> = "policy violation".into();
let err = UrlTargetError::PolicyRejected(source);
assert!(
std::error::Error::source(&err).is_some(),
"PolicyRejected should preserve source for programmatic access"
);
}
}