server/lib/src/pull_intake/http.rs
//! **The pull-through's one outbound door.** A blocking GET with bounded
//! timeouts, a byte cap, the offline guard, and redirects followed by hand.
//!
//! * Every dial asks [`crate::egress::check`] first, so a `network: Offline`
//! Holger refuses before a socket exists.
//! * A redirect is followed at most [`MAX_HOPS`] times. A hop to another origin
//! must be `https` (registries hand blobs to a CDN, PyPI's files live on
//! another host); a downgrade from `https` to `http` is refused.
//! * The source's credential goes ONLY to the source's own origin, never to the
//! host a redirect names.
//! * Nothing fetched here is trusted for what it says: every file is held
//! against the digest its index or manifest publishes before it reaches the
//! chain, and the chain then scans it like an upload.
#[cfg(feature = "online")]
use std::sync::Arc;
/// Redirect hops a fetch follows.
pub const MAX_HOPS: usize = 4;
/// What one GET answered.
#[derive(Debug, Clone, Default)]
pub struct Reply {
pub status: u16,
/// Lower-case header names.
pub headers: Vec<(String, String)>,
pub body: Vec<u8>,
/// The URL that answered (after redirects).
pub url: String,
}
impl Reply {
pub fn header(&self, name: &str) -> Option<&str> {
let name = name.to_ascii_lowercase();
self.headers
.iter()
.find(|(k, _)| *k == name)
.map(|(_, v)| v.as_str())
}
}
/// **A blocking HTTP GET.** The ureq client in the `online` build; the tests'
/// fakes are real HTTP servers behind the same client.
pub trait Fetch: Send + Sync {
/// GET `url` with `headers`. `auth` is sent only while the request stays on
/// `auth_origin`. A 3xx is followed here; any other status is returned.
fn get(
&self,
url: &str,
headers: &[(String, String)],
auth: Option<(&str, &str)>,
cap: u64,
) -> anyhow::Result<Reply>;
}
/// `scheme://host:port` of an absolute URL, lower-cased; `None` for anything else.
pub fn origin_of(url: &str) -> Option<String> {
let (scheme, rest) = url.split_once("://")?;
let scheme = scheme.to_ascii_lowercase();
if scheme != "http" && scheme != "https" {
return None;
}
let authority = rest
.split(['/', '?', '#'])
.next()
.filter(|a| !a.is_empty())?;
if authority.contains('@') {
return None;
}
let default_port = if scheme == "https" { 443 } else { 80 };
let (host, port) = match authority.rsplit_once(':') {
Some((h, p)) if !h.ends_with(']') || authority.starts_with('[') => match p.parse::<u16>() {
Ok(p) => (h.to_ascii_lowercase(), p),
Err(_) => (authority.to_ascii_lowercase(), default_port),
},
_ => (authority.to_ascii_lowercase(), default_port),
};
Some(format!("{scheme}://{host}:{port}"))
}
/// Resolve `href` (absolute, protocol-relative, absolute-path or relative) against `base`.
pub fn join(base: &str, href: &str) -> Option<String> {
let href = href.trim();
if href.contains(['\r', '\n', ' ']) {
return None;
}
if href.starts_with("http://") || href.starts_with("https://") {
return Some(href.to_string());
}
let (scheme, rest) = base.split_once("://")?;
let authority = rest.split(['/', '?', '#']).next()?;
if let Some(hostpath) = href.strip_prefix("//") {
return Some(format!("{scheme}://{hostpath}"));
}
if href.starts_with('/') {
return Some(format!("{scheme}://{authority}{href}"));
}
// Relative: drop the base's query and last segment, then resolve `..` and `.`.
let path = &rest[authority.len()..];
let path = path.split(['?', '#']).next().unwrap_or("");
let dir = match path.rfind('/') {
Some(i) => &path[..=i],
None => "/",
};
let mut segs: Vec<&str> = dir.split('/').filter(|s| !s.is_empty()).collect();
let (href_path, tail) = match href.find(['?', '#']) {
Some(i) => (&href[..i], &href[i..]),
None => (href, ""),
};
let ends_slash = href_path.ends_with('/') || href_path.is_empty();
for s in href_path.split('/') {
match s {
"" | "." => {}
".." => {
segs.pop();
}
s => segs.push(s),
}
}
let mut out = format!("{scheme}://{authority}/{}", segs.join("/"));
if ends_slash && !segs.is_empty() && !out.ends_with('/') {
out.push('/');
}
out.push_str(tail);
Some(out)
}
/// The ureq-backed [`Fetch`] (`online` build).
#[cfg(feature = "online")]
pub struct UreqFetch {
agent: ureq::Agent,
}
#[cfg(feature = "online")]
impl UreqFetch {
pub fn new() -> Self {
Self {
agent: crate::upstream_rust::build_agent(
std::time::Duration::from_secs(10),
std::time::Duration::from_secs(60),
),
}
}
pub fn shared() -> Arc<dyn Fetch> {
static ONE: std::sync::OnceLock<Arc<UreqFetch>> = std::sync::OnceLock::new();
let f: Arc<UreqFetch> = Arc::clone(ONE.get_or_init(|| Arc::new(UreqFetch::new())));
f
}
}
#[cfg(feature = "online")]
impl Default for UreqFetch {
fn default() -> Self {
Self::new()
}
}
#[cfg(feature = "online")]
impl Fetch for UreqFetch {
fn get(
&self,
url: &str,
headers: &[(String, String)],
auth: Option<(&str, &str)>,
cap: u64,
) -> anyhow::Result<Reply> {
use std::io::Read as _;
let auth_origin = auth.and_then(|(o, _)| origin_of(o));
let mut current = url.to_string();
let first_https = url.starts_with("https://");
for _ in 0..=MAX_HOPS {
crate::egress::check("pull-through", ¤t)?;
let here = origin_of(¤t)
.ok_or_else(|| anyhow::anyhow!("not an absolute http(s) url: {current}"))?;
let mut req = self.agent.get(¤t);
for (k, v) in headers {
req = req.header(k.as_str(), v.as_str());
}
if let (Some(o), Some((_, value))) = (&auth_origin, auth) {
if *o == here {
req = req.header("Authorization", value);
}
}
let resp = req
.call()
.map_err(|e| anyhow::anyhow!("GET {current}: {e}"))?;
let status = resp.status().as_u16();
if (300..400).contains(&status) && status != 304 {
let location = resp
.headers()
.get("location")
.and_then(|v| v.to_str().ok())
.ok_or_else(|| {
anyhow::anyhow!("GET {current}: HTTP {status} without a Location")
})?
.to_string();
let next = join(¤t, &location).ok_or_else(|| {
anyhow::anyhow!("GET {current}: unusable Location '{location}'")
})?;
let next_origin = origin_of(&next)
.ok_or_else(|| anyhow::anyhow!("GET {current}: redirect to '{next}'"))?;
if next_origin != here && !next.starts_with("https://") {
anyhow::bail!("GET {current}: refusing a redirect to another origin over plain http ({next})");
}
if first_https && !next.starts_with("https://") {
anyhow::bail!("GET {current}: refusing a downgrade from https to {next}");
}
current = next;
continue;
}
let hdrs: Vec<(String, String)> = resp
.headers()
.iter()
.filter_map(|(k, v)| {
Some((
k.as_str().to_ascii_lowercase(),
v.to_str().ok()?.to_string(),
))
})
.collect();
let mut body = Vec::new();
resp.into_body()
.into_reader()
.take(cap.saturating_add(1))
.read_to_end(&mut body)
.map_err(|e| anyhow::anyhow!("GET {current}: reading the body: {e}"))?;
if body.len() as u64 > cap {
anyhow::bail!("GET {current}: the body is larger than the {cap}-byte cap");
}
return Ok(Reply {
status,
headers: hdrs,
body,
url: current,
});
}
anyhow::bail!("GET {url}: more than {MAX_HOPS} redirects")
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn join_resolves_every_href_shape() {
let base = "https://pypi.org/simple/sampleproject/";
assert_eq!(
join(base, "https://files.example/x.whl").unwrap(),
"https://files.example/x.whl"
);
assert_eq!(
join(base, "//cdn.example/x").unwrap(),
"https://cdn.example/x"
);
assert_eq!(
join(base, "/packages/x.whl").unwrap(),
"https://pypi.org/packages/x.whl"
);
assert_eq!(
join(base, "../../packages/x.whl#sha256=ab").unwrap(),
"https://pypi.org/packages/x.whl#sha256=ab"
);
assert_eq!(
join("http://127.0.0.1:9/simple/p/", "x.whl").unwrap(),
"http://127.0.0.1:9/simple/p/x.whl"
);
}
#[test]
fn origin_carries_scheme_host_and_port() {
assert_eq!(
origin_of("https://cgr.dev/v2/").unwrap(),
"https://cgr.dev:443"
);
assert_eq!(
origin_of("http://127.0.0.1:8080/x").unwrap(),
"http://127.0.0.1:8080"
);
assert!(origin_of("https://u@evil/").is_none());
assert!(origin_of("file:///etc/passwd").is_none());
}
}