use crate::error::{DepsError, Result};
use crate::net_policy::{RegistryAccessPolicy, WorkspaceRegistryAccess};
use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
use reqwest::{Client, Response, StatusCode, Url, header};
use serde::Serialize;
use std::borrow::Cow;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Instant;
const MAX_CACHE_ENTRIES: usize = 1000;
const MAX_CACHE_BYTES: usize = 64 * 1024 * 1024;
const MAX_CACHEABLE_ENTRY_BYTES: usize = MAX_CACHE_BYTES / 8;
const HTTP_TIMEOUT_SECS: u64 = 30;
const MAX_RESPONSE_BYTES: usize = 32 * 1024 * 1024;
const ABSOLUTE_MAX_RESPONSE_BYTES: usize = 128 * 1024 * 1024;
const CACHE_EVICTION_PERCENTAGE: usize = 10;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BodyLimit(usize);
impl BodyLimit {
pub const DEFAULT: Self = Self(MAX_RESPONSE_BYTES);
#[must_use]
pub const fn new(bytes: usize) -> Self {
if bytes > ABSOLUTE_MAX_RESPONSE_BYTES {
Self(ABSOLUTE_MAX_RESPONSE_BYTES)
} else {
Self(bytes)
}
}
#[must_use]
pub const fn bytes(self) -> usize {
self.0
}
}
#[cfg(any(test, feature = "test-util"))]
fn is_loopback_host(url: &str) -> bool {
let Some(rest) = url
.strip_prefix("http://")
.or_else(|| url.strip_prefix("https://"))
else {
return false;
};
let host_and_port = rest.split(['/', '?', '#']).next().unwrap_or("");
let host = if let Some(bracketed) = host_and_port.strip_prefix('[') {
bracketed.split(']').next().unwrap_or("")
} else {
host_and_port.split(':').next().unwrap_or("")
};
matches!(host, "127.0.0.1" | "localhost" | "::1")
}
#[inline]
fn ensure_https(url: &str) -> Result<()> {
if url.starts_with("https://") {
return Ok(());
}
#[cfg(any(test, feature = "test-util"))]
if is_loopback_host(url) {
return Ok(());
}
Err(DepsError::CacheError(format!("URL must use HTTPS: {url}")))
}
fn is_https_downgrade(previous: &Url, next: &Url) -> bool {
previous.scheme() == "https" && next.scheme() == "http"
}
fn hop_targets_blocked_host(url: &Url) -> bool {
let class = crate::net_policy::classify_host(url);
#[cfg(any(test, feature = "test-util"))]
let class_blocked = class.never_a_registry() && class != crate::net_policy::HostClass::Loopback;
#[cfg(not(any(test, feature = "test-util")))]
let class_blocked = class.never_a_registry();
class_blocked
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CacheTier {
Baseline,
WorkspaceDeclared(WorkspaceRegistryAccess),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AddrGuard {
Baseline,
WorkspaceDeclared(WorkspaceRegistryAccess),
}
impl AddrGuard {
fn tier_allows(self, class: crate::net_policy::HostClass) -> bool {
match self {
Self::Baseline => true,
Self::WorkspaceDeclared(policy) => policy.allows(class),
}
}
}
fn redirect_policy(guard: AddrGuard) -> reqwest::redirect::Policy {
reqwest::redirect::Policy::custom(move |attempt| {
let downgraded = attempt
.previous()
.last()
.is_some_and(|previous| is_https_downgrade(previous, attempt.url()));
if downgraded
|| hop_targets_blocked_host(attempt.url())
|| !guard.tier_allows(crate::net_policy::classify_host(attempt.url()))
{
attempt.stop()
} else {
reqwest::redirect::Policy::default().redirect(attempt)
}
})
}
fn trusted_origin_redirect_policy(trusted_origin: String) -> reqwest::redirect::Policy {
reqwest::redirect::Policy::custom(move |attempt| {
if attempt.url().as_str().starts_with(&trusted_origin) {
reqwest::redirect::Policy::default().redirect(attempt)
} else {
attempt.stop()
}
})
}
#[derive(Debug, thiserror::Error)]
enum ResolveGuardError {
#[error("DNS resolution for {host} returned no addresses")]
NoAddresses { host: String },
#[error("resolved address {addr} for host {host} is {class}, blocked by net_policy")]
Blocked {
host: String,
addr: std::net::IpAddr,
class: crate::net_policy::HostClass,
},
}
fn validate_resolved_addrs(
host: &str,
addrs: Vec<std::net::SocketAddr>,
guard: AddrGuard,
) -> std::result::Result<Vec<std::net::SocketAddr>, ResolveGuardError> {
if addrs.is_empty() {
tracing::warn!(host, "DNS resolution returned no addresses");
return Err(ResolveGuardError::NoAddresses {
host: host.to_string(),
});
}
for addr in &addrs {
let class = crate::net_policy::classify_addr(addr.ip());
if class.never_a_registry() || !guard.tier_allows(class) {
tracing::warn!(host, addr = %addr.ip(), %class, "blocking DNS-resolved address");
return Err(ResolveGuardError::Blocked {
host: host.to_string(),
addr: addr.ip(),
class,
});
}
}
Ok(addrs)
}
#[cfg(test)]
type SyntheticLookupFn = dyn Fn(&str) -> Vec<std::net::SocketAddr> + Send + Sync;
#[cfg(test)]
#[derive(Clone)]
struct TestLookup(Arc<SyntheticLookupFn>);
#[cfg(test)]
impl std::fmt::Debug for TestLookup {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("TestLookup(..)")
}
}
#[derive(Debug, Clone)]
struct BlockedAddrResolver {
guard: AddrGuard,
#[cfg(test)]
lookup: Option<TestLookup>,
}
impl BlockedAddrResolver {
fn new(guard: AddrGuard) -> Self {
Self {
guard,
#[cfg(test)]
lookup: None,
}
}
#[cfg(test)]
fn with_lookup(guard: AddrGuard, lookup: TestLookup) -> Self {
Self {
guard,
lookup: Some(lookup),
}
}
}
impl reqwest::dns::Resolve for BlockedAddrResolver {
fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
let host = name.as_str().to_string();
let guard = self.guard;
#[cfg(test)]
let lookup = self.lookup.clone();
Box::pin(async move {
#[cfg(test)]
let addrs: Vec<std::net::SocketAddr> = match &lookup {
Some(lookup) => (lookup.0)(&host),
None => tokio::net::lookup_host((host.as_str(), 0)).await?.collect(),
};
#[cfg(not(test))]
let addrs: Vec<std::net::SocketAddr> =
tokio::net::lookup_host((host.as_str(), 0)).await?.collect();
let addrs = validate_resolved_addrs(&host, addrs, guard)?;
Ok(Box::new(addrs.into_iter()) as reqwest::dns::Addrs)
})
}
}
fn build_client_inner(
redirect: reqwest::redirect::Policy,
resolver: BlockedAddrResolver,
) -> Client {
Client::builder()
.user_agent(format!("deps-lsp/{}", env!("CARGO_PKG_VERSION")))
.timeout(std::time::Duration::from_secs(HTTP_TIMEOUT_SECS))
.redirect(redirect)
.dns_resolver(resolver)
.build()
.expect("failed to create HTTP client")
}
fn build_guarded_client(guard: AddrGuard) -> Client {
build_client_inner(redirect_policy(guard), BlockedAddrResolver::new(guard))
}
#[cfg(test)]
fn build_guarded_client_with_lookup(guard: AddrGuard, lookup: TestLookup) -> Client {
build_client_inner(
redirect_policy(guard),
BlockedAddrResolver::with_lookup(guard, lookup),
)
}
#[derive(Clone)]
struct Transport {
client: Client,
tier: CacheTier,
}
impl Transport {
fn baseline() -> Self {
Self {
client: build_guarded_client(AddrGuard::Baseline),
tier: CacheTier::Baseline,
}
}
fn workspace(policy: &Arc<RegistryAccessPolicy>) -> Self {
let snapshot = policy.get();
Self {
client: build_guarded_client(AddrGuard::WorkspaceDeclared(snapshot)),
tier: CacheTier::WorkspaceDeclared(snapshot),
}
}
fn origin_pinned(trusted_origin: &str) -> Self {
Self {
client: build_client_inner(
trusted_origin_redirect_policy(trusted_origin.to_string()),
BlockedAddrResolver::new(AddrGuard::Baseline),
),
tier: CacheTier::Baseline,
}
}
}
async fn read_body_capped(url: &str, mut response: Response, limit: BodyLimit) -> Result<Bytes> {
let mut body = BytesMut::new();
let limit = limit.bytes();
while let Some(chunk) = response
.chunk()
.await
.map_err(|e| DepsError::RegistryError {
package: url.to_string(),
source: e,
})?
{
if body.len() + chunk.len() > limit {
return Err(DepsError::ResponseTooLarge {
url: url.to_string(),
limit,
});
}
body.extend_from_slice(&chunk);
}
Ok(body.freeze())
}
#[derive(Debug, Clone)]
pub struct CachedResponse {
pub body: Bytes,
pub etag: Option<String>,
pub last_modified: Option<String>,
pub fetched_at: Instant,
}
pub struct HttpCache {
entries: DashMap<String, CachedResponse>,
total_bytes: AtomicUsize,
baseline: Transport,
trusted_clients: DashMap<String, Transport>,
policy: Arc<RegistryAccessPolicy>,
workspace: RwLock<Transport>,
#[cfg(test)]
workspace_rebuilds: AtomicUsize,
offline: AtomicBool,
cache_enabled: AtomicBool,
}
impl HttpCache {
pub fn new() -> Self {
Self::with_policy(Arc::new(RegistryAccessPolicy::default()))
}
pub fn with_policy(policy: Arc<RegistryAccessPolicy>) -> Self {
let workspace = Transport::workspace(&policy);
Self {
entries: DashMap::new(),
total_bytes: AtomicUsize::new(0),
baseline: Transport::baseline(),
trusted_clients: DashMap::new(),
policy,
workspace: RwLock::new(workspace),
#[cfg(test)]
workspace_rebuilds: AtomicUsize::new(0),
offline: AtomicBool::new(false),
cache_enabled: AtomicBool::new(true),
}
}
pub fn set_offline(&self, value: bool) {
self.offline.store(value, Ordering::Relaxed);
}
#[must_use]
pub fn is_offline(&self) -> bool {
self.offline.load(Ordering::Relaxed)
}
pub fn set_cache_enabled(&self, value: bool) {
self.cache_enabled.store(value, Ordering::Relaxed);
}
fn ensure_online(&self, url: &str) -> Result<()> {
if self.is_offline() {
return Err(DepsError::Offline {
url: url.to_string(),
});
}
Ok(())
}
fn transport_for_origin(&self, trusted_origin: &str) -> Transport {
if let Some(existing) = self.trusted_clients.get(trusted_origin) {
return existing.clone();
}
self.trusted_clients
.entry(trusted_origin.to_string())
.or_insert_with(|| Transport::origin_pinned(trusted_origin))
.clone()
}
const WS_KEY_PREFIX: char = '\u{1}';
fn cache_key<'a>(&self, url: &'a str, tier: CacheTier) -> Cow<'a, str> {
match tier {
CacheTier::Baseline => Cow::Borrowed(url),
CacheTier::WorkspaceDeclared(snapshot) => {
Cow::Owned(format!("{}{}{url}", Self::WS_KEY_PREFIX, snapshot.to_u8()))
}
}
}
pub async fn get_cached(&self, url: &str) -> Result<Bytes> {
self.get_cached_with_headers(url, &[]).await
}
#[must_use]
pub fn peek_cached(&self, url: &str) -> Option<Bytes> {
self.entries.get(url).map(|r| r.body.clone())
}
pub async fn get_cached_with_headers(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
) -> Result<Bytes> {
self.get_cached_with_headers_via(url, extra_headers, &self.baseline)
.await
}
pub async fn get_cached_trusted_origin(
&self,
url: &str,
trusted_origin: &str,
) -> Result<Bytes> {
let transport = self.transport_for_origin(trusted_origin);
self.get_cached_with_headers_via(url, &[], &transport).await
}
pub async fn get_cached_trusted_origin_with_headers(
&self,
url: &str,
trusted_origin: &str,
extra_headers: &[(header::HeaderName, &str)],
) -> Result<Bytes> {
let transport = self.transport_for_origin(trusted_origin);
self.get_cached_with_headers_via(url, extra_headers, &transport)
.await
}
pub async fn get_cached_workspace(&self, url: &str) -> Result<Bytes> {
self.get_cached_workspace_with_headers(url, &[]).await
}
pub async fn get_cached_workspace_with_headers(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
) -> Result<Bytes> {
let transport = self
.workspace
.read()
.expect("workspace transport lock poisoned")
.clone();
self.get_cached_with_headers_via(url, extra_headers, &transport)
.await
}
pub fn set_registry_policy(&self, value: WorkspaceRegistryAccess) {
if self.policy.get() == value {
return;
}
self.policy.set(value);
let rebuilt = Transport::workspace(&self.policy);
*self
.workspace
.write()
.expect("workspace transport lock poisoned") = rebuilt;
#[cfg(test)]
self.workspace_rebuilds.fetch_add(1, Ordering::Relaxed);
}
async fn get_cached_with_headers_via(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
transport: &Transport,
) -> Result<Bytes> {
if self.entries.len() >= MAX_CACHE_ENTRIES
|| self.total_bytes.load(Ordering::Relaxed) >= MAX_CACHE_BYTES
{
self.evict_entries();
}
let offline = self.is_offline();
let cache_enabled = self.cache_enabled.load(Ordering::Relaxed) || offline;
let cache_key = self.cache_key(url, transport.tier);
if !cache_enabled {
return self
.transport_only_via(url, extra_headers, BodyLimit::DEFAULT, &transport.client)
.await;
}
if let Some(cached) = self.entries.get(cache_key.as_ref()).map(|r| r.clone()) {
if offline {
return Ok(cached.body);
}
match self
.conditional_request_with_headers(
url,
&cached,
extra_headers,
&transport.client,
&cache_key,
)
.await
{
Ok(Some(new_body)) => return Ok(new_body),
Ok(None) => return Ok(cached.body),
Err(e) => {
tracing::warn!("conditional request failed, using cache: {e}");
return Ok(cached.body);
}
}
}
self.fetch_and_store_with_headers(url, extra_headers, &transport.client, &cache_key)
.await
}
async fn conditional_request_with_headers(
&self,
url: &str,
cached: &CachedResponse,
extra_headers: &[(header::HeaderName, &str)],
client: &Client,
cache_key: &str,
) -> Result<Option<Bytes>> {
self.ensure_online(url)?;
ensure_https(url)?;
let mut request = client.get(url);
for (name, value) in extra_headers {
request = request.header(name, *value);
}
if let Some(etag) = &cached.etag {
request = request.header(header::IF_NONE_MATCH, etag);
}
if let Some(last_modified) = &cached.last_modified {
request = request.header(header::IF_MODIFIED_SINCE, last_modified);
}
let response = request.send().await.map_err(|e| DepsError::RegistryError {
package: url.to_string(),
source: e,
})?;
if response.status() == StatusCode::NOT_MODIFIED {
return Ok(None);
}
if !response.status().is_success() {
return Err(DepsError::HttpStatus {
url: url.to_string(),
status: response.status().as_u16(),
});
}
let etag = response
.headers()
.get(header::ETAG)
.and_then(|v| v.to_str().ok())
.map(String::from);
let last_modified = response
.headers()
.get(header::LAST_MODIFIED)
.and_then(|v| v.to_str().ok())
.map(String::from);
let body = read_body_capped(url, response, BodyLimit::DEFAULT).await?;
self.store_entry(
cache_key.to_string(),
CachedResponse {
body: body.clone(),
etag,
last_modified,
fetched_at: Instant::now(),
},
);
Ok(Some(body))
}
async fn fetch_and_store_with_headers(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
client: &Client,
cache_key: &str,
) -> Result<Bytes> {
self.ensure_online(url)?;
ensure_https(url)?;
tracing::debug!(extra_headers = extra_headers.len(), "fetching fresh: {url}");
let mut request = client.get(url);
for (name, value) in extra_headers {
request = request.header(name, *value);
}
let response = request.send().await.map_err(|e| DepsError::RegistryError {
package: url.to_string(),
source: e,
})?;
if !response.status().is_success() {
return Err(DepsError::HttpStatus {
url: url.to_string(),
status: response.status().as_u16(),
});
}
let etag = response
.headers()
.get(header::ETAG)
.and_then(|v| v.to_str().ok())
.map(String::from);
let last_modified = response
.headers()
.get(header::LAST_MODIFIED)
.and_then(|v| v.to_str().ok())
.map(String::from);
let body = read_body_capped(url, response, BodyLimit::DEFAULT).await?;
self.store_entry(
cache_key.to_string(),
CachedResponse {
body: body.clone(),
etag,
last_modified,
fetched_at: Instant::now(),
},
);
Ok(body)
}
pub async fn post_json<T: Serialize + ?Sized>(&self, url: &str, body: &T) -> Result<Bytes> {
self.ensure_online(url)?;
ensure_https(url)?;
let response = self
.baseline
.client
.post(url)
.json(body)
.send()
.await
.map_err(|e| DepsError::RegistryError {
package: url.to_string(),
source: e,
})?;
if !response.status().is_success() {
return Err(DepsError::HttpStatus {
url: url.to_string(),
status: response.status().as_u16(),
});
}
read_body_capped(url, response, BodyLimit::DEFAULT).await
}
pub async fn get_transport_only(&self, url: &str) -> Result<Bytes> {
self.get_transport_only_with_headers(url, &[]).await
}
pub async fn get_transport_only_with_headers(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
) -> Result<Bytes> {
self.get_transport_only_with_headers_limited(url, extra_headers, BodyLimit::DEFAULT)
.await
}
pub async fn get_transport_only_with_headers_limited(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
limit: BodyLimit,
) -> Result<Bytes> {
self.transport_only_via(url, extra_headers, limit, &self.baseline.client)
.await
}
pub async fn get_transport_only_with_headers_limited_trusted_origin(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
limit: BodyLimit,
trusted_origin: &str,
) -> Result<Bytes> {
let transport = self.transport_for_origin(trusted_origin);
self.transport_only_via(url, extra_headers, limit, &transport.client)
.await
}
async fn transport_only_via(
&self,
url: &str,
extra_headers: &[(header::HeaderName, &str)],
limit: BodyLimit,
client: &Client,
) -> Result<Bytes> {
self.ensure_online(url)?;
ensure_https(url)?;
let mut request = client.get(url);
for (name, value) in extra_headers {
request = request.header(name, *value);
}
let response = request.send().await.map_err(|e| DepsError::RegistryError {
package: url.to_string(),
source: e,
})?;
if !response.status().is_success() {
return Err(DepsError::HttpStatus {
url: url.to_string(),
status: response.status().as_u16(),
});
}
read_body_capped(url, response, limit).await
}
fn store_entry(&self, url: String, response: CachedResponse) {
let new_len = response.body.len();
if new_len > MAX_CACHEABLE_ENTRY_BYTES {
if let Some((_, old)) = self.entries.remove(&url) {
self.total_bytes
.fetch_sub(old.body.len(), Ordering::Relaxed);
}
return;
}
let old_len = self
.entries
.insert(url, response)
.map_or(0, |old| old.body.len());
self.total_bytes.fetch_add(new_len, Ordering::Relaxed);
self.total_bytes.fetch_sub(old_len, Ordering::Relaxed);
}
pub fn clear(&self) {
self.entries.clear();
self.total_bytes.store(0, Ordering::Relaxed);
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn total_bytes(&self) -> usize {
self.total_bytes.load(Ordering::Relaxed)
}
fn evict_entries(&self) {
use std::cmp::Reverse;
use std::collections::BinaryHeap;
let count_target_removals = if self.entries.len() >= MAX_CACHE_ENTRIES {
(MAX_CACHE_ENTRIES / CACHE_EVICTION_PERCENTAGE).max(1)
} else {
0
};
let mut oldest: BinaryHeap<Reverse<(Instant, String)>> = self
.entries
.iter()
.map(|entry| Reverse((entry.value().fetched_at, entry.key().clone())))
.collect();
let mut removed = 0usize;
while removed < count_target_removals
|| self.total_bytes.load(Ordering::Relaxed) > MAX_CACHE_BYTES
{
let Some(Reverse((_, url))) = oldest.pop() else {
break;
};
if let Some((_, old)) = self.entries.remove(&url) {
self.total_bytes
.fetch_sub(old.body.len(), Ordering::Relaxed);
}
removed += 1;
}
tracing::debug!(
"evicted {removed} cache entries ({} bytes remaining)",
self.total_bytes.load(Ordering::Relaxed)
);
}
#[doc(hidden)]
pub fn get_for_bench(&self, url: &str) -> Option<Bytes> {
self.entries.get(url).map(|entry| entry.body.clone())
}
#[doc(hidden)]
pub fn insert_for_bench(&self, url: String, response: CachedResponse) {
self.store_entry(url, response);
}
}
impl Default for HttpCache {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_ensure_https_rejects_non_loopback_http() {
assert!(ensure_https("http://example.com").is_err());
}
#[test]
fn test_ensure_https_rejects_hosts_resembling_loopback() {
assert!(ensure_https("http://localhost.evil.com/").is_err());
assert!(ensure_https("http://127.0.0.1.evil.com/").is_err());
assert!(ensure_https("http://evil.com/?cb=127.0.0.1").is_err());
}
#[test]
fn test_ensure_https_accepts_bracketed_ipv6_loopback() {
assert!(ensure_https("http://[::1]:1234/x").is_ok());
}
#[test]
fn test_is_https_downgrade() {
let https = Url::parse("https://example.com/a").unwrap();
let http = Url::parse("http://example.com/a").unwrap();
assert!(is_https_downgrade(&https, &http));
assert!(!is_https_downgrade(&http, &https));
assert!(!is_https_downgrade(&https, &https));
assert!(!is_https_downgrade(&http, &http));
}
#[test]
fn test_hop_targets_blocked_host_blocks_cloud_metadata() {
let url = Url::parse("https://169.254.169.254/latest/meta-data/").unwrap();
assert!(hop_targets_blocked_host(&url));
}
#[test]
fn test_hop_targets_blocked_host_allows_global() {
let url = Url::parse("https://index.crates.io/").unwrap();
assert!(!hop_targets_blocked_host(&url));
}
#[test]
fn test_hop_targets_blocked_host_exempts_loopback_under_test_cfg() {
let url = Url::parse("http://127.0.0.1:1234/api/target").unwrap();
assert!(!hop_targets_blocked_host(&url));
}
#[test]
fn test_validate_resolved_addrs_blocks_cloud_metadata() {
let addrs = vec!["169.254.169.254:0".parse().unwrap()];
assert!(matches!(
validate_resolved_addrs("evil.example", addrs, AddrGuard::Baseline),
Err(ResolveGuardError::Blocked { .. })
));
}
#[test]
fn test_validate_resolved_addrs_blocks_when_any_address_is_blocked() {
let addrs = vec![
"1.1.1.1:0".parse().unwrap(),
"169.254.169.254:0".parse().unwrap(),
];
assert!(matches!(
validate_resolved_addrs("evil.example", addrs, AddrGuard::Baseline),
Err(ResolveGuardError::Blocked { .. })
));
}
#[test]
fn test_validate_resolved_addrs_allows_global() {
let addrs = vec!["1.1.1.1:0".parse().unwrap()];
assert_eq!(
validate_resolved_addrs("index.crates.io", addrs.clone(), AddrGuard::Baseline).unwrap(),
addrs
);
}
#[test]
fn test_validate_resolved_addrs_fails_closed_on_empty() {
assert!(matches!(
validate_resolved_addrs("evil.example", vec![], AddrGuard::Baseline),
Err(ResolveGuardError::NoAddresses { .. })
));
}
#[test]
fn test_validate_resolved_addrs_unwraps_mapped_v4() {
let addrs = vec!["[::ffff:169.254.169.254]:0".parse().unwrap()];
assert!(matches!(
validate_resolved_addrs("evil.example", addrs, AddrGuard::Baseline),
Err(ResolveGuardError::Blocked { .. })
));
}
#[test]
fn test_validate_resolved_addrs_baseline_allows_private_ranges() {
for addr_str in ["10.0.0.1:0", "100.64.0.1:0", "[fc00::1]:0"] {
let addrs = vec![addr_str.parse().unwrap()];
assert!(
validate_resolved_addrs("corp.example", addrs, AddrGuard::Baseline).is_ok(),
"{addr_str} must be allowed under Baseline"
);
}
}
#[test]
fn test_validate_resolved_addrs_workspace_public_only_blocks_private_ranges() {
let guard = AddrGuard::WorkspaceDeclared(WorkspaceRegistryAccess::PublicOnly);
for addr_str in ["10.0.0.1:0", "100.64.0.1:0", "[fc00::1]:0"] {
let addrs = vec![addr_str.parse().unwrap()];
assert!(
matches!(
validate_resolved_addrs("evil.example", addrs, guard),
Err(ResolveGuardError::Blocked { .. })
),
"{addr_str} must be blocked under WorkspaceDeclared(PublicOnly)"
);
}
let global = vec!["1.1.1.1:0".parse().unwrap()];
assert!(validate_resolved_addrs("index.crates.io", global, guard).is_ok());
}
#[test]
fn test_validate_resolved_addrs_workspace_all_allows_private_ranges() {
let guard = AddrGuard::WorkspaceDeclared(WorkspaceRegistryAccess::All);
let addrs = vec!["10.0.0.1:0".parse().unwrap()];
assert!(validate_resolved_addrs("corp.example", addrs, guard).is_ok());
}
#[test]
fn test_validate_resolved_addrs_workspace_off_rejects_global() {
let guard = AddrGuard::WorkspaceDeclared(WorkspaceRegistryAccess::Off);
let addrs = vec!["1.1.1.1:0".parse().unwrap()];
assert!(matches!(
validate_resolved_addrs("index.crates.io", addrs, guard),
Err(ResolveGuardError::Blocked { .. })
));
}
#[tokio::test]
async fn test_build_guarded_client_with_lookup_blocks_private_range_under_workspace_tier() {
let lookup = TestLookup(Arc::new(|_host: &str| vec!["10.0.0.1:0".parse().unwrap()]));
let guard = AddrGuard::WorkspaceDeclared(WorkspaceRegistryAccess::PublicOnly);
let client = build_guarded_client_with_lookup(guard, lookup);
let result = client.get("https://corp.example/").send().await;
let err = result.expect_err(
"a private-range synthetic lookup must be rejected under WorkspaceDeclared(PublicOnly)",
);
assert!(
format!("{err:?}").contains("Blocked"),
"expected rejection at the resolver-guard step, got: {err:?}"
);
}
#[tokio::test]
async fn test_blocked_addr_resolver_allows_private_range_under_baseline() {
use reqwest::dns::Resolve;
let lookup = TestLookup(Arc::new(|_host: &str| vec!["10.0.0.1:0".parse().unwrap()]));
let resolver = BlockedAddrResolver::with_lookup(AddrGuard::Baseline, lookup);
let result = resolver.resolve("corp.example".parse().unwrap()).await;
assert!(
result.is_ok(),
"Baseline must allow a private-range address through"
);
}
#[tokio::test]
async fn test_blocked_addr_resolver_rejects_loopback_name_directly() {
use reqwest::dns::Resolve;
let addrs = BlockedAddrResolver::new(AddrGuard::Baseline)
.resolve("localhost".parse().unwrap())
.await;
assert!(addrs.is_err());
}
#[tokio::test]
async fn test_build_client_wires_in_blocked_addr_resolver() {
let mut server = mockito::Server::new_async().await;
let _mock = server
.mock("GET", "/")
.with_status(200)
.create_async()
.await;
let port = server.socket_address().port();
let client = build_guarded_client(AddrGuard::Baseline);
let result = client.get(format!("http://localhost:{port}/")).send().await;
let err = result.expect_err(
"expected the wired-in resolver guard to reject a loopback-resolving name even \
though a real listener answers at this port",
);
let debug = format!("{err:?}");
assert!(
debug.contains("Blocked") && debug.contains("Loopback"),
"expected the failure to originate from ResolveGuardError::Blocked with class \
Loopback, got: {debug}"
);
}
#[tokio::test]
async fn test_get_cached_stops_redirect_to_cloud_metadata() {
let mut server = mockito::Server::new_async().await;
let _redirect = server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", "https://169.254.169.254/latest/meta-data/")
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", server.url());
let result: Result<Bytes> = cache.get_cached(&source_url).await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected the redirect to be stopped and surfaced as HttpStatus(302)"
);
}
#[tokio::test]
async fn test_get_cached_follows_same_scheme_redirect() {
let mut server = mockito::Server::new_async().await;
let target_url = format!("{}/api/target", server.url());
let _redirect = server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &target_url)
.create_async()
.await;
let _target = server
.mock("GET", "/api/target")
.with_status(200)
.with_body("redirected data")
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", server.url());
let result: Bytes = cache.get_cached(&source_url).await.unwrap();
assert_eq!(result.as_ref(), b"redirected data");
}
#[tokio::test]
async fn test_workspace_transport_stops_loopback_redirect_baseline_follows() {
let mut server_a = mockito::Server::new_async().await;
let mut server_b = mockito::Server::new_async().await;
let target_url = format!("{}/api/target", server_b.url());
let _redirect = server_a
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &target_url)
.create_async()
.await;
let _target = server_b
.mock("GET", "/api/target")
.with_status(200)
.with_body("redirected data")
.create_async()
.await;
let source_url = format!("{}/api/source", server_a.url());
let cache = HttpCache::new();
let result: Bytes = cache
.get_cached_with_headers_via(&source_url, &[], &Transport::baseline())
.await
.unwrap();
assert_eq!(result.as_ref(), b"redirected data");
let policy = Arc::new(RegistryAccessPolicy::new(
WorkspaceRegistryAccess::PublicOnly,
));
let workspace_cache = HttpCache::with_policy(Arc::clone(&policy));
let result: Result<Bytes> = workspace_cache
.get_cached_with_headers_via(&source_url, &[], &Transport::workspace(&policy))
.await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected the workspace transport to stop the loopback hop, got {result:?}"
);
}
#[tokio::test]
async fn test_workspace_transport_stops_redirect_to_private_literal() {
let mut server = mockito::Server::new_async().await;
let _redirect = server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", "https://10.0.0.1/x")
.create_async()
.await;
let policy = Arc::new(RegistryAccessPolicy::new(
WorkspaceRegistryAccess::PublicOnly,
));
let cache = HttpCache::with_policy(Arc::clone(&policy));
let source_url = format!("{}/api/source", server.url());
let result: Result<Bytes> = cache
.get_cached_with_headers_via(&source_url, &[], &Transport::workspace(&policy))
.await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected the redirect to 10.0.0.1 to be stopped, got {result:?}"
);
}
#[tokio::test]
async fn test_get_cached_trusted_origin_stops_cross_origin_redirect() {
let mut trusted_server = mockito::Server::new_async().await;
let mut other_server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/", trusted_server.url());
let escape_target = format!("{}/api/stolen", other_server.url());
let _redirect = trusted_server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &escape_target)
.create_async()
.await;
let escape = other_server
.mock("GET", "/api/stolen")
.with_status(200)
.with_body("must not be returned")
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", trusted_server.url());
let result: Result<Bytes> = cache
.get_cached_trusted_origin(&source_url, &trusted_origin)
.await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected HttpStatus(302)"
);
escape.assert_async().await;
}
#[tokio::test]
async fn test_get_cached_trusted_origin_follows_same_origin_redirect() {
let mut server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/", server.url());
let target_url = format!("{}/api/target", server.url());
let _redirect = server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &target_url)
.create_async()
.await;
let _target = server
.mock("GET", "/api/target")
.with_status(200)
.with_body("trusted data")
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", server.url());
let result: Bytes = cache
.get_cached_trusted_origin(&source_url, &trusted_origin)
.await
.unwrap();
assert_eq!(result.as_ref(), b"trusted data");
}
#[tokio::test]
async fn test_get_cached_trusted_origin_stops_second_hop_of_multi_hop_chain() {
let mut trusted_server = mockito::Server::new_async().await;
let mut other_server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/", trusted_server.url());
let intermediate_url = format!("{}/api/intermediate", trusted_server.url());
let escape_target = format!("{}/api/stolen", other_server.url());
let _first_hop = trusted_server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &intermediate_url)
.create_async()
.await;
let _second_hop = trusted_server
.mock("GET", "/api/intermediate")
.with_status(302)
.with_header("location", &escape_target)
.create_async()
.await;
let escape = other_server
.mock("GET", "/api/stolen")
.with_status(200)
.with_body("must not be returned")
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", trusted_server.url());
let result: Result<Bytes> = cache
.get_cached_trusted_origin(&source_url, &trusted_origin)
.await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected HttpStatus(302)"
);
escape.assert_async().await;
}
#[tokio::test]
async fn test_get_cached_trusted_origin_rejects_sibling_path_prefix() {
let mut server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/api/", server.url());
let escape_target = format!("{}/apiX/evil", server.url());
let _redirect = server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &escape_target)
.create_async()
.await;
let escape = server
.mock("GET", "/apiX/evil")
.with_status(200)
.with_body("must not be returned")
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", server.url());
let result: Result<Bytes> = cache
.get_cached_trusted_origin(&source_url, &trusted_origin)
.await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected HttpStatus(302)"
);
escape.assert_async().await;
}
#[tokio::test]
async fn test_get_cached_trusted_origin_with_headers_sends_extra_header() {
let mut server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/", server.url());
let _m = server
.mock("GET", "/api/data")
.match_header("authorization", "Bearer secret-token")
.with_status(200)
.with_body("authenticated data")
.create_async()
.await;
let cache = HttpCache::new();
let url = format!("{}/api/data", server.url());
let result: Bytes = cache
.get_cached_trusted_origin_with_headers(
&url,
&trusted_origin,
&[(header::AUTHORIZATION, "Bearer secret-token")],
)
.await
.unwrap();
assert_eq!(result.as_ref(), b"authenticated data");
}
#[tokio::test]
async fn test_get_cached_trusted_origin_with_headers_stops_cross_origin_redirect() {
let mut trusted_server = mockito::Server::new_async().await;
let mut other_server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/", trusted_server.url());
let escape_target = format!("{}/api/stolen", other_server.url());
let _redirect = trusted_server
.mock("GET", "/api/source")
.with_status(302)
.with_header("location", &escape_target)
.create_async()
.await;
let escape = other_server
.mock("GET", "/api/stolen")
.with_status(200)
.with_body("must not be returned")
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
let source_url = format!("{}/api/source", trusted_server.url());
let result: Result<Bytes> = cache
.get_cached_trusted_origin_with_headers(
&source_url,
&trusted_origin,
&[(header::AUTHORIZATION, "Bearer secret-token")],
)
.await;
assert!(
matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
"expected HttpStatus(302)"
);
escape.assert_async().await;
}
#[tokio::test]
async fn test_get_cached_default_client_enforces_ten_hop_redirect_limit() {
let mut server = mockito::Server::new_async().await;
let base = server.url();
let mut hop_mocks = Vec::new();
for i in 0..11u32 {
let path = format!("/step/{i}");
let next = format!("{base}/step/{}", i + 1);
hop_mocks.push(
server
.mock("GET", path.as_str())
.with_status(302)
.with_header("location", &next)
.create_async()
.await,
);
}
let final_step = server
.mock("GET", "/step/11")
.with_status(200)
.with_body("unreachable")
.expect(0)
.create_async()
.await;
assert_eq!(hop_mocks.len(), 11);
let cache = HttpCache::new();
let start_url = format!("{base}/step/0");
let result: Result<Bytes> = cache.get_cached(&start_url).await;
assert!(
matches!(result, Err(DepsError::RegistryError { .. })),
"expected a too-many-redirects network error, got {result:?}"
);
final_step.assert_async().await;
}
#[test]
fn test_cache_creation() {
let cache = HttpCache::new();
assert_eq!(cache.len(), 0);
assert!(cache.is_empty());
}
#[test]
fn test_cache_clear() {
let cache = HttpCache::new();
cache.entries.insert(
"test".into(),
CachedResponse {
body: Bytes::from_static(&[1, 2, 3]),
etag: None,
last_modified: None,
fetched_at: Instant::now(),
},
);
assert_eq!(cache.len(), 1);
cache.clear();
assert_eq!(cache.len(), 0);
}
#[test]
fn test_cached_response_clone() {
let response = CachedResponse {
body: Bytes::from_static(&[1, 2, 3]),
etag: Some("test".into()),
last_modified: Some("date".into()),
fetched_at: Instant::now(),
};
let cloned = response.clone();
assert_eq!(response.body, cloned.body);
assert_eq!(response.etag, cloned.etag);
}
#[test]
fn test_cache_len() {
let cache = HttpCache::new();
assert_eq!(cache.len(), 0);
cache.entries.insert(
"url1".into(),
CachedResponse {
body: Bytes::new(),
etag: None,
last_modified: None,
fetched_at: Instant::now(),
},
);
assert_eq!(cache.len(), 1);
}
#[tokio::test]
async fn test_get_cached_fresh_fetch() {
let mut server = mockito::Server::new_async().await;
let _m = server
.mock("GET", "/api/data")
.with_status(200)
.with_header("etag", "\"abc123\"")
.with_body("test data")
.create_async()
.await;
let cache = HttpCache::new();
let url = format!("{}/api/data", server.url());
let result: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result.as_ref(), b"test data");
assert_eq!(cache.len(), 1);
}
#[tokio::test]
async fn test_get_cached_cache_hit() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let cache = HttpCache::new();
let _m1 = server
.mock("GET", "/api/data")
.with_status(200)
.with_header("etag", "\"abc123\"")
.with_body("original data")
.create_async()
.await;
let result1: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result1.as_ref(), b"original data");
assert_eq!(cache.len(), 1);
drop(_m1);
let _m2 = server
.mock("GET", "/api/data")
.match_header("if-none-match", "\"abc123\"")
.with_status(304)
.create_async()
.await;
let result2: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result2.as_ref(), b"original data");
}
#[tokio::test]
async fn test_get_cached_304_not_modified() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let cache = HttpCache::new();
let _m1 = server
.mock("GET", "/api/data")
.with_status(200)
.with_header("etag", "\"abc123\"")
.with_body("original data")
.create_async()
.await;
let result1: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result1.as_ref(), b"original data");
drop(_m1);
let _m2 = server
.mock("GET", "/api/data")
.match_header("if-none-match", "\"abc123\"")
.with_status(304)
.create_async()
.await;
let result2: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result2.as_ref(), b"original data");
}
#[tokio::test]
async fn test_get_cached_etag_validation() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let cache = HttpCache::new();
cache.entries.insert(
url.clone(),
CachedResponse {
body: Bytes::from_static(b"cached"),
etag: Some("\"tag123\"".into()),
last_modified: None,
fetched_at: Instant::now(),
},
);
let _m = server
.mock("GET", "/api/data")
.match_header("if-none-match", "\"tag123\"")
.with_status(304)
.create_async()
.await;
let result: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result.as_ref(), b"cached");
}
#[tokio::test]
async fn test_get_cached_last_modified_validation() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let cache = HttpCache::new();
cache.entries.insert(
url.clone(),
CachedResponse {
body: Bytes::from_static(b"cached"),
etag: None,
last_modified: Some("Wed, 21 Oct 2024 07:28:00 GMT".into()),
fetched_at: Instant::now(),
},
);
let _m = server
.mock("GET", "/api/data")
.match_header("if-modified-since", "Wed, 21 Oct 2024 07:28:00 GMT")
.with_status(304)
.create_async()
.await;
let result: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result.as_ref(), b"cached");
}
#[tokio::test]
async fn test_get_cached_network_error_fallback() {
let cache = HttpCache::new();
let url = "https://invalid.localhost.test/data";
cache.entries.insert(
url.to_string(),
CachedResponse {
body: Bytes::from_static(b"stale data"),
etag: Some("\"old\"".into()),
last_modified: None,
fetched_at: Instant::now(),
},
);
let result: Bytes = cache.get_cached(url).await.unwrap();
assert_eq!(result.as_ref(), b"stale data");
}
#[tokio::test]
async fn test_fetch_and_store_http_error() {
let mut server = mockito::Server::new_async().await;
let _m = server
.mock("GET", "/api/missing")
.with_status(404)
.with_body("Not Found")
.create_async()
.await;
let cache = HttpCache::new();
let url = format!("{}/api/missing", server.url());
let result: Result<Bytes> = cache
.fetch_and_store_with_headers(&url, &[], &cache.baseline.client, &url)
.await;
assert!(result.is_err());
match result {
Err(DepsError::HttpStatus { status, .. }) => {
assert_eq!(status, 404);
}
_ => panic!("Expected HttpStatus"),
}
}
#[tokio::test]
async fn test_fetch_and_store_stores_headers() {
let mut server = mockito::Server::new_async().await;
let _m = server
.mock("GET", "/api/data")
.with_status(200)
.with_header("etag", "\"abc123\"")
.with_header("last-modified", "Wed, 21 Oct 2024 07:28:00 GMT")
.with_body("test")
.create_async()
.await;
let cache = HttpCache::new();
let url = format!("{}/api/data", server.url());
let _: Bytes = cache
.fetch_and_store_with_headers(&url, &[], &cache.baseline.client, &url)
.await
.unwrap();
let cached = cache.entries.get(&url).unwrap();
assert_eq!(cached.etag, Some("\"abc123\"".into()));
assert_eq!(
cached.last_modified,
Some("Wed, 21 Oct 2024 07:28:00 GMT".into())
);
}
#[tokio::test]
async fn test_get_cached_with_headers_sends_extra_headers() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let _m = server
.mock("GET", "/api/data")
.match_header("authorization", "Bearer token123")
.with_status(200)
.with_header("etag", "\"abc123\"")
.with_body("authed data")
.create_async()
.await;
let cache = HttpCache::new();
let headers = [(header::AUTHORIZATION, "Bearer token123")];
let result: Bytes = cache.get_cached_with_headers(&url, &headers).await.unwrap();
assert_eq!(result.as_ref(), b"authed data");
}
#[tokio::test]
async fn test_get_cached_workspace_with_headers_sends_extra_headers() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let _m = server
.mock("GET", "/api/data")
.match_header("accept", "application/vnd.npm.install-v1+json")
.with_status(200)
.with_body("abbreviated packument")
.create_async()
.await;
let cache = HttpCache::new();
let headers = [(header::ACCEPT, "application/vnd.npm.install-v1+json")];
let result: Bytes = cache
.get_cached_workspace_with_headers(&url, &headers)
.await
.unwrap();
assert_eq!(result.as_ref(), b"abbreviated packument");
}
#[tokio::test]
async fn test_fetch_and_store_rejects_oversized_response() {
let mut server = mockito::Server::new_async().await;
let oversized_body = vec![0u8; MAX_RESPONSE_BYTES + 1];
let _m = server
.mock("GET", "/api/huge")
.with_status(200)
.with_body(oversized_body)
.create_async()
.await;
let cache = HttpCache::new();
let url = format!("{}/api/huge", server.url());
let result: Result<Bytes> = cache
.fetch_and_store_with_headers(&url, &[], &cache.baseline.client, &url)
.await;
match result {
Err(DepsError::ResponseTooLarge { limit, .. }) => {
assert_eq!(limit, MAX_RESPONSE_BYTES);
}
other => panic!("expected ResponseTooLarge, got {other:?}"),
}
assert!(cache.entries.get(&url).is_none());
}
#[tokio::test]
async fn test_fetch_and_store_accepts_response_at_exact_cap() {
let mut server = mockito::Server::new_async().await;
let exact_cap_body = vec![0u8; MAX_RESPONSE_BYTES];
let _m = server
.mock("GET", "/api/exact")
.with_status(200)
.with_body(exact_cap_body)
.create_async()
.await;
let cache = HttpCache::new();
let url = format!("{}/api/exact", server.url());
let result: Bytes = cache
.fetch_and_store_with_headers(&url, &[], &cache.baseline.client, &url)
.await
.unwrap();
assert_eq!(result.len(), MAX_RESPONSE_BYTES);
}
#[tokio::test]
async fn test_get_cached_non_2xx_on_refresh_preserves_stale_cache() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let cache = HttpCache::new();
cache.entries.insert(
url.clone(),
CachedResponse {
body: Bytes::from_static(b"stale but good"),
etag: Some("\"stale-etag\"".into()),
last_modified: None,
fetched_at: Instant::now(),
},
);
let _m = server
.mock("GET", "/api/data")
.match_header("if-none-match", "\"stale-etag\"")
.with_status(503)
.with_body("<html>maintenance</html>")
.create_async()
.await;
let result: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result.as_ref(), b"stale but good");
let cached = cache.entries.get(&url).unwrap();
assert_eq!(cached.etag, Some("\"stale-etag\"".into()));
}
#[tokio::test]
async fn test_post_json_success_returns_body_and_does_not_cache() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/querybatch", server.url());
let _m = server
.mock("POST", "/v1/querybatch")
.match_header("content-type", "application/json")
.with_status(200)
.with_body(r#"{"results":[{}]}"#)
.create_async()
.await;
let cache = HttpCache::new();
let body = serde_json::json!({ "queries": [] });
let result: Bytes = cache.post_json(&url, &body).await.unwrap();
assert_eq!(result.as_ref(), br#"{"results":[{}]}"#);
assert!(
cache.is_empty(),
"post_json must not populate the entry-map cache"
);
}
#[tokio::test]
async fn test_post_json_non_2xx_returns_http_status_error() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/querybatch", server.url());
let _m = server
.mock("POST", "/v1/querybatch")
.with_status(400)
.create_async()
.await;
let cache = HttpCache::new();
let body = serde_json::json!({ "queries": [] });
let result: Result<Bytes> = cache.post_json(&url, &body).await;
match result {
Err(DepsError::HttpStatus { status, .. }) => assert_eq!(status, 400),
other => panic!("expected HttpStatus, got {other:?}"),
}
}
#[tokio::test]
async fn test_get_transport_only_success_returns_body_and_does_not_cache() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/vulns/RUSTSEC-2020-0071", server.url());
let _m = server
.mock("GET", "/v1/vulns/RUSTSEC-2020-0071")
.with_status(200)
.with_body(r#"{"id":"RUSTSEC-2020-0071"}"#)
.create_async()
.await;
let cache = HttpCache::new();
let result: Bytes = cache.get_transport_only(&url).await.unwrap();
assert_eq!(result.as_ref(), br#"{"id":"RUSTSEC-2020-0071"}"#);
assert!(
cache.is_empty(),
"get_transport_only must not populate the entry-map cache"
);
}
#[tokio::test]
async fn test_get_transport_only_with_headers_sends_extra_headers() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/vulns/RUSTSEC-2020-0071", server.url());
let _m = server
.mock("GET", "/v1/vulns/RUSTSEC-2020-0071")
.match_header("accept", "application/json")
.with_status(200)
.with_body(r#"{"id":"RUSTSEC-2020-0071"}"#)
.create_async()
.await;
let cache = HttpCache::new();
let headers = [(header::ACCEPT, "application/json")];
let result: Bytes = cache
.get_transport_only_with_headers(&url, &headers)
.await
.unwrap();
assert_eq!(result.as_ref(), br#"{"id":"RUSTSEC-2020-0071"}"#);
assert!(
cache.is_empty(),
"get_transport_only_with_headers must not populate the entry-map cache"
);
}
#[tokio::test]
async fn test_get_transport_only_non_2xx_returns_http_status_error() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/vulns/missing", server.url());
let _m = server
.mock("GET", "/v1/vulns/missing")
.with_status(404)
.create_async()
.await;
let cache = HttpCache::new();
let result: Result<Bytes> = cache.get_transport_only(&url).await;
match result {
Err(DepsError::HttpStatus { status, .. }) => assert_eq!(status, 404),
other => panic!("expected HttpStatus, got {other:?}"),
}
}
fn dummy_response(size: usize) -> CachedResponse {
CachedResponse {
body: Bytes::from(vec![0u8; size]),
etag: None,
last_modified: None,
fetched_at: Instant::now(),
}
}
#[test]
fn test_total_bytes_tracks_inserts_and_replacement() {
let cache = HttpCache::new();
cache.store_entry("url1".into(), dummy_response(100));
assert_eq!(cache.total_bytes(), 100);
cache.store_entry("url1".into(), dummy_response(40));
assert_eq!(cache.total_bytes(), 40);
cache.store_entry("url2".into(), dummy_response(60));
assert_eq!(cache.total_bytes(), 100);
}
#[test]
fn test_clear_resets_total_bytes() {
let cache = HttpCache::new();
cache.store_entry("url1".into(), dummy_response(1000));
assert_eq!(cache.total_bytes(), 1000);
cache.clear();
assert_eq!(cache.total_bytes(), 0);
}
#[test]
fn test_small_payloads_do_not_trigger_eviction() {
let cache = HttpCache::new();
for i in 0..50 {
cache.store_entry(format!("url{i}"), dummy_response(1024));
}
assert_eq!(cache.len(), 50);
assert_eq!(cache.total_bytes(), 50 * 1024);
}
#[test]
fn test_evict_entries_triggers_on_byte_budget_with_few_entries() {
let cache = HttpCache::new();
for i in 0..9 {
cache.store_entry(format!("url{i}"), dummy_response(MAX_CACHEABLE_ENTRY_BYTES));
}
assert_eq!(cache.len(), 9);
assert!(cache.total_bytes() > MAX_CACHE_BYTES);
cache.evict_entries();
assert!(cache.total_bytes() <= MAX_CACHE_BYTES);
assert_eq!(cache.len(), 8);
}
#[test]
fn test_evict_entries_removes_oldest_first_for_bytes() {
let cache = HttpCache::new();
cache.store_entry("oldest".into(), dummy_response(MAX_CACHEABLE_ENTRY_BYTES));
std::thread::sleep(std::time::Duration::from_millis(5));
for i in 0..8 {
cache.store_entry(
format!("newer{i}"),
dummy_response(MAX_CACHEABLE_ENTRY_BYTES),
);
}
assert_eq!(cache.len(), 9);
cache.evict_entries();
assert_eq!(cache.len(), 8);
assert!(cache.entries.get("oldest").is_none());
for i in 0..8 {
assert!(cache.entries.get(&format!("newer{i}")).is_some());
}
}
#[tokio::test]
async fn test_get_cached_with_headers_evicts_on_byte_budget() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let cache = HttpCache::new();
for i in 0..9 {
cache.store_entry(
format!("stale{i}"),
dummy_response(MAX_CACHEABLE_ENTRY_BYTES),
);
}
assert!(cache.total_bytes() > MAX_CACHE_BYTES);
let _m = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("fresh")
.create_async()
.await;
let result: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result.as_ref(), b"fresh");
assert!(cache.total_bytes() <= MAX_CACHE_BYTES + result.len());
}
#[test]
fn test_store_entry_skips_caching_oversized_entry() {
let cache = HttpCache::new();
cache.store_entry("big".into(), dummy_response(MAX_CACHEABLE_ENTRY_BYTES + 1));
assert!(cache.entries.get("big").is_none());
assert_eq!(cache.total_bytes(), 0);
cache.store_entry("small".into(), dummy_response(100));
assert_eq!(cache.total_bytes(), 100);
cache.store_entry(
"small".into(),
dummy_response(MAX_CACHEABLE_ENTRY_BYTES + 1),
);
assert!(cache.entries.get("small").is_none());
assert_eq!(cache.total_bytes(), 0);
}
#[test]
fn test_concurrent_store_and_evict_keeps_total_bytes_consistent() {
use std::sync::Arc;
use std::thread;
let cache = Arc::new(HttpCache::new());
let handles: Vec<_> = (0..8)
.map(|t| {
let cache = Arc::clone(&cache);
thread::spawn(move || {
for i in 0..150 {
cache.store_entry(format!("t{t}-{i}"), dummy_response(4096));
if i % 10 == 0 {
cache.evict_entries();
}
}
})
})
.collect();
for handle in handles {
handle.join().unwrap();
}
cache.evict_entries();
let actual: usize = cache
.entries
.iter()
.map(|entry| entry.value().body.len())
.sum();
assert_eq!(cache.total_bytes(), actual);
}
#[tokio::test]
async fn test_get_cached_and_get_cached_workspace_do_not_share_an_entry() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("shared url, distinct tiers")
.expect(2)
.create_async()
.await;
let cache = HttpCache::new();
let baseline_result: Bytes = cache.get_cached(&url).await.unwrap();
let workspace_result: Bytes = cache.get_cached_workspace(&url).await.unwrap();
assert_eq!(baseline_result.as_ref(), b"shared url, distinct tiers");
assert_eq!(workspace_result.as_ref(), b"shared url, distinct tiers");
assert_eq!(cache.len(), 2);
mock.assert_async().await;
}
#[tokio::test]
async fn test_set_registry_policy_change_does_not_serve_stale_era_body() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let _first = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("all-era body")
.create_async()
.await;
let policy = Arc::new(RegistryAccessPolicy::new(WorkspaceRegistryAccess::All));
let cache = HttpCache::with_policy(Arc::clone(&policy));
let first: Bytes = cache.get_cached_workspace(&url).await.unwrap();
assert_eq!(first.as_ref(), b"all-era body");
cache.set_registry_policy(WorkspaceRegistryAccess::PublicOnly);
let _second = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("public-only-era body")
.create_async()
.await;
let second: Bytes = cache.get_cached_workspace(&url).await.unwrap();
assert_eq!(
second.as_ref(),
b"public-only-era body",
"the All-era cached body must not be served after tightening to PublicOnly"
);
assert_eq!(cache.len(), 2);
}
#[test]
fn test_set_registry_policy_rebuilds_only_on_change() {
let policy = Arc::new(RegistryAccessPolicy::new(
WorkspaceRegistryAccess::PublicOnly,
));
let cache = HttpCache::with_policy(policy);
assert_eq!(cache.workspace_rebuilds.load(Ordering::Relaxed), 0);
cache.set_registry_policy(WorkspaceRegistryAccess::PublicOnly);
assert_eq!(
cache.workspace_rebuilds.load(Ordering::Relaxed),
0,
"re-applying the unchanged policy must not rebuild the workspace transport"
);
cache.set_registry_policy(WorkspaceRegistryAccess::All);
assert_eq!(cache.workspace_rebuilds.load(Ordering::Relaxed), 1);
cache.set_registry_policy(WorkspaceRegistryAccess::All);
assert_eq!(
cache.workspace_rebuilds.load(Ordering::Relaxed),
1,
"re-applying the unchanged (new) policy must not rebuild again"
);
cache.set_registry_policy(WorkspaceRegistryAccess::Off);
assert_eq!(cache.workspace_rebuilds.load(Ordering::Relaxed), 2);
}
#[tokio::test]
async fn test_offline_cold_get_cached_errors_without_network() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("must not be fetched")
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
cache.set_offline(true);
let result: Result<Bytes> = cache.get_cached(&url).await;
match result {
Err(DepsError::Offline { url: blocked }) => assert_eq!(blocked, url),
other => panic!("expected Offline, got {other:?}"),
}
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_cold_post_json_errors_without_network() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/querybatch", server.url());
let mock = server
.mock("POST", "/v1/querybatch")
.with_status(200)
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
cache.set_offline(true);
let body = serde_json::json!({ "queries": [] });
let result: Result<Bytes> = cache.post_json(&url, &body).await;
assert!(matches!(result, Err(DepsError::Offline { .. })));
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_cold_get_transport_only_errors_without_network() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/v1/vulns/RUSTSEC-2020-0071", server.url());
let mock = server
.mock("GET", "/v1/vulns/RUSTSEC-2020-0071")
.with_status(200)
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
cache.set_offline(true);
let result: Result<Bytes> = cache.get_transport_only(&url).await;
assert!(matches!(result, Err(DepsError::Offline { .. })));
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_warm_serves_cached_body_without_network() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("must not be fetched")
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
cache.entries.insert(
url.clone(),
CachedResponse {
body: Bytes::from_static(b"warm cached body"),
etag: Some("\"tag123\"".into()),
last_modified: None,
fetched_at: Instant::now(),
},
);
cache.set_offline(true);
let result: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(result.as_ref(), b"warm cached body");
mock.assert_async().await;
}
#[tokio::test]
async fn test_cache_disabled_two_calls_each_hit_the_server() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("fresh every time")
.expect(2)
.create_async()
.await;
let cache = HttpCache::new();
cache.set_cache_enabled(false);
let first: Bytes = cache.get_cached(&url).await.unwrap();
let second: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(first.as_ref(), b"fresh every time");
assert_eq!(second.as_ref(), b"fresh every time");
assert!(
cache.is_empty(),
"cache.enabled: false must never populate the entry map"
);
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_and_cache_disabled_cold_start_errors_cleanly() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.expect(0)
.create_async()
.await;
let cache = HttpCache::new();
cache.set_cache_enabled(false);
cache.set_offline(true);
let result: Result<Bytes> = cache.get_cached(&url).await;
assert!(matches!(result, Err(DepsError::Offline { .. })));
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_overrides_disabled_cache_to_serve_warm_entry() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("fetched while online")
.expect(1)
.create_async()
.await;
let cache = HttpCache::new();
let first: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(first.as_ref(), b"fetched while online");
cache.set_cache_enabled(false);
cache.set_offline(true);
let second: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(
second.as_ref(),
b"fetched while online",
"offline must override cache_enabled:false and still serve the warm entry"
);
mock.assert_async().await;
}
#[tokio::test]
async fn test_online_to_offline_transition_serves_previously_fetched_entry() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("fetched while online")
.expect(1)
.create_async()
.await;
let cache = HttpCache::new();
let online: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(online.as_ref(), b"fetched while online");
cache.set_offline(true);
let offline: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(offline.as_ref(), b"fetched while online");
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_to_online_transition_resumes_fetching() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_header("etag", "\"abc123\"")
.with_body("fetched while online")
.expect(1)
.create_async()
.await;
let cache = HttpCache::new();
let online: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(online.as_ref(), b"fetched while online");
mock.assert_async().await;
cache.set_offline(true);
let offline: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(offline.as_ref(), b"fetched while online");
cache.set_offline(false);
drop(mock);
let revalidate = server
.mock("GET", "/api/data")
.match_header("if-none-match", "\"abc123\"")
.with_status(304)
.expect(1)
.create_async()
.await;
let resumed: Bytes = cache.get_cached(&url).await.unwrap();
assert_eq!(
resumed.as_ref(),
b"fetched while online",
"returning online must resume live requests, not stay pinned to the cached body"
);
revalidate.assert_async().await;
}
#[tokio::test]
async fn test_offline_warm_serves_workspace_tier_without_network() {
let mut server = mockito::Server::new_async().await;
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("workspace fetch")
.expect(1)
.create_async()
.await;
let cache = HttpCache::new();
let online: Bytes = cache.get_cached_workspace(&url).await.unwrap();
assert_eq!(online.as_ref(), b"workspace fetch");
cache.set_offline(true);
let offline: Bytes = cache.get_cached_workspace(&url).await.unwrap();
assert_eq!(offline.as_ref(), b"workspace fetch");
mock.assert_async().await;
}
#[tokio::test]
async fn test_offline_warm_serves_trusted_origin_tier_without_network() {
let mut server = mockito::Server::new_async().await;
let trusted_origin = format!("{}/", server.url());
let url = format!("{}/api/data", server.url());
let mock = server
.mock("GET", "/api/data")
.with_status(200)
.with_body("trusted-origin fetch")
.expect(1)
.create_async()
.await;
let cache = HttpCache::new();
let online: Bytes = cache
.get_cached_trusted_origin(&url, &trusted_origin)
.await
.unwrap();
assert_eq!(online.as_ref(), b"trusted-origin fetch");
cache.set_offline(true);
let offline: Bytes = cache
.get_cached_trusted_origin(&url, &trusted_origin)
.await
.unwrap();
assert_eq!(offline.as_ref(), b"trusted-origin fetch");
mock.assert_async().await;
}
}