deps-core 1.1.0

//! HTTP response cache shared by every ecosystem's registry client.
//!
//! Wraps outbound registry requests with RFC 7232 conditional-request
//! validation (`ETag`/`If-None-Match`, `Last-Modified`/`If-Modified-Since`) so
//! that unchanged registry data is served from a bounded in-memory cache
//! instead of re-fetched. Entry count and total retained bytes are both
//! capped to keep memory use predictable under long-running LSP sessions.

use crate::cache_policy::CACHE_EVICTION_PERCENTAGE;
use crate::error::{DepsError, Result};
use crate::net_policy::{RedactedUrl, 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::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Instant;

/// Maximum number of cached entries to prevent unbounded memory growth.
const MAX_CACHE_ENTRIES: usize = 1000;

/// Maximum total bytes retained across all cached response bodies.
///
/// `MAX_CACHE_ENTRIES` alone bounds entry *count*, not size: since a single
/// response body may be as large as [`MAX_RESPONSE_BYTES`] (32 MiB), a cache
/// full of near-cap entries could retain tens of gigabytes even though real
/// registry payloads are typically well under 1 MB. This budget is a
/// defense-in-depth cap (CWE-400) against that worst case, evicted
/// alongside the count-based limit in [`HttpCache::evict_entries`]. 64 MiB
/// comfortably holds thousands of typical registry responses while still
/// bounding the pathological case.
///
/// This is a best-effort bound, not a hard guarantee: it is checked once
/// per request in [`HttpCache::get_cached_with_headers_via`], so multiple
/// requests already in flight when the budget is crossed can each finish
/// inserting before the next check fires. [`MAX_CACHEABLE_ENTRY_BYTES`]
/// keeps that per-request overshoot small (at most one admission-cap-sized
/// insert per concurrent in-flight request) rather than bounding it exactly.
const MAX_CACHE_BYTES: usize = 64 * 1024 * 1024;

/// Maximum size of a single response body that will be retained in the
/// cache; larger bodies are still returned to the caller, just never
/// stored.
///
/// Without this cap, [`MAX_CACHE_BYTES`] alone lets a handful of
/// large-but-legitimate responses (up to [`MAX_RESPONSE_BYTES`], 32 MiB
/// each) evict the *entire* rest of the cache: at a 2x ratio between the
/// two constants, just two max-size entries would saturate the whole
/// budget. Set to an eighth of [`MAX_CACHE_BYTES`] (8 MiB) so no single
/// entry can claim more than 1/8 of the budget — a handful of large
/// responses degrade to "not cached" instead of "evicts the small-payload
/// working set".
const MAX_CACHEABLE_ENTRY_BYTES: usize = MAX_CACHE_BYTES / 8;

/// HTTP request timeout in seconds.
const HTTP_TIMEOUT_SECS: u64 = 30;

/// Maximum decompressed response body size accepted from a single request.
///
/// `reqwest`'s `gzip` feature strips `Content-Length`/`Content-Encoding` after
/// decoding a response, so a header-based pre-check cannot bound body size
/// (`response.content_length()` is `None` for every decoded response). This
/// cap is instead enforced by counting bytes as the body streams in, aborting
/// as soon as the running total would exceed the limit.
const MAX_RESPONSE_BYTES: usize = 32 * 1024 * 1024;

/// Ceiling every [`BodyLimit`] is clamped to at construction, so no caller can weaken
/// the size guard [`read_body_capped`] enforces past this value.
///
/// 128 MiB comfortably covers the largest known caller (`deps-pypi`'s PyPI
/// Simple API full index, ~43 MB decompressed today, capped at 96 MiB for organic
/// growth) while still bounding the pathological case.
const ABSOLUTE_MAX_RESPONSE_BYTES: usize = 128 * 1024 * 1024;

/// Upper bound on a single response body, clamped at construction so no caller can
/// weaken the guard `read_body_capped` enforces past `ABSOLUTE_MAX_RESPONSE_BYTES`.
///
/// Every cache method that previously read `MAX_RESPONSE_BYTES` directly now takes
/// this newtype instead (defaulting to it via [`Self::DEFAULT`]), so a caller that
/// legitimately needs a larger cap — e.g. a full-index fetch that bypasses the entry
/// cache entirely, like [`HttpCache::get_transport_only_with_headers_limited`] — can
/// request one without touching the shared constant every other registry client
/// relies on.
///
/// # Examples
///
/// ```
/// use deps_core::cache::BodyLimit;
///
/// let default_limit = BodyLimit::DEFAULT;
/// let clamped = BodyLimit::new(usize::MAX);
/// assert_ne!(clamped, default_limit);
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BodyLimit(usize);

impl BodyLimit {
    /// The default limit (`MAX_RESPONSE_BYTES`), used by every cache method that
    /// does not take an explicit [`BodyLimit`].
    pub const DEFAULT: Self = Self(MAX_RESPONSE_BYTES);

    /// Creates a limit of `bytes`, clamped down to `ABSOLUTE_MAX_RESPONSE_BYTES` if
    /// `bytes` exceeds it.
    #[must_use]
    pub const fn new(bytes: usize) -> Self {
        if bytes > ABSOLUTE_MAX_RESPONSE_BYTES {
            Self(ABSOLUTE_MAX_RESPONSE_BYTES)
        } else {
            Self(bytes)
        }
    }

    /// The clamped byte value this limit enforces.
    #[must_use]
    pub const fn bytes(self) -> usize {
        self.0
    }
}

/// Whether `url`'s host is loopback (`127.0.0.1`, `localhost`, or `::1`), with any scheme
/// and an optional port — the shape every `mockito::Server` binds to.
///
/// Only compiled into test builds (see [`ensure_https`]): a non-loopback host must never
/// be allowed to bypass the HTTPS requirement, even under `cfg(test)`/`test-util`. See
/// [`crate::net_policy::validate_index_url`]'s own private loopback check for the
/// counterpart used on parsed `url::Url` values — kept separate rather than merged, since
/// this one takes a raw `&str` on `ensure_https`'s hot path and has looser (any-scheme)
/// semantics.
#[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")
}

/// Validates that a URL uses HTTPS protocol.
///
/// Returns an error if the URL doesn't start with "https://".
/// This ensures all network requests are encrypted.
///
/// A loopback HTTP URL (`127.0.0.1`/`localhost`/`::1`, the shape every `mockito::Server`
/// binds to) is allowed in `deps-core`'s own test builds (`cfg(test)`) and in other
/// workspace crates' test builds via the `test-util` feature — `cfg(test)` alone does not
/// apply there, since those crates depend on `deps-core` as a normal, non-dev dependency.
/// Any other HTTP host is still rejected even under those cfgs: `test-util` is a public,
/// independently-enablable crates.io feature, so this must not become "any host, any
/// environment" just because the feature is on.
#[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: {}",
        RedactedUrl::new(url)
    )))
}

/// True when a redirect hop moves from an `https` origin to a plain `http` one.
///
/// A redirect to any scheme other than `http`/`https` is already rejected by reqwest
/// itself once a hop is followed, so the downgrade case is the only one this needs to
/// catch here.
fn is_https_downgrade(previous: &Url, next: &Url) -> bool {
    previous.scheme() == "https" && next.scheme() == "http"
}

/// Whether a redirect hop's target host is one [`crate::net_policy::HostClass::never_a_registry`]
/// blocks, exempting `Loopback` in test builds — the identical carve-out [`ensure_https`]
/// already uses, without which every mockito redirect chain in this workspace's tests would
/// break.
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
}

/// Which cache-key namespace and [`AddrGuard`] tier a [`Transport`] enforces.
///
/// `Baseline` is every non-workspace request (all 11 ecosystems' registry/redirect/API
/// traffic); `WorkspaceDeclared` is Cargo's workspace-declared-registry traffic, carrying the
/// same [`WorkspaceRegistryAccess`] **value snapshot** an [`AddrGuard::WorkspaceDeclared`]
/// carries (see that variant's docs) — [`HttpCache::cache_key`] reads the digit from this
/// snapshot, never from a live `Arc<RegistryAccessPolicy>` read, so a request's cache key and
/// its guard always agree on which policy era they were constructed under.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum CacheTier {
    Baseline,
    WorkspaceDeclared(WorkspaceRegistryAccess),
    /// An origin-pinned, connect-address-guarded tier (issue #561/#562) — see
    /// [`Transport::origin_pinned_guarded`]. `digest` identifies the `(trusted_origin,
    /// policy_snapshot)` pair this transport was built for (**never** a credential identity —
    /// see [`HttpCache::get_cached_pinned_with_headers`]'s separate `auth_id` argument, folded
    /// into the cache key only). `authenticated` distinguishes an authenticated fetch (#561)
    /// from #562's unauthenticated workspace-declared one for cache-eviction purposes
    /// ([`CacheTier::is_authenticated`]) without affecting pooling — the shipped
    /// [`Transport::origin_pinned`] (public path) stays on [`CacheTier::Baseline`], unaffected.
    Pinned {
        digest: u64,
        authenticated: bool,
    },
}

impl CacheTier {
    /// Whether a 401/403 revalidation response on an entry under this tier should evict the
    /// entry rather than serve the default stale-while-revalidate fallback (FR-015/NFR-004).
    fn is_authenticated(self) -> bool {
        matches!(
            self,
            Self::Pinned {
                authenticated: true,
                ..
            }
        )
    }
}

/// The tier a [`Transport`]'s redirect policy and DNS resolver both enforce.
///
/// `WorkspaceDeclared` holds a [`WorkspaceRegistryAccess`] **value snapshot**, taken once at
/// [`Transport::workspace`] construction time — not a live `Arc<RegistryAccessPolicy>` read on
/// every [`Self::tier_allows`] call. [`Transport::workspace`] takes this same snapshot for its
/// paired [`CacheTier::WorkspaceDeclared`], and [`HttpCache::set_registry_policy`] rebuilds the
/// whole `Transport` (both snapshots included) on every actual policy transition, so a
/// request's cache key and its guard always come from one consistent construction-time value,
/// with no read-skew window between them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AddrGuard {
    Baseline,
    WorkspaceDeclared(WorkspaceRegistryAccess),
}

impl AddrGuard {
    /// Whether a host classified as `class` may be reached under this guard's tier.
    fn tier_allows(self, class: crate::net_policy::HostClass) -> bool {
        match self {
            Self::Baseline => true,
            Self::WorkspaceDeclared(policy) => policy.allows(class),
        }
    }
}

/// Redirect policy for a [`Transport`]'s client, parameterized by the [`AddrGuard`] tier that
/// client enforces.
///
/// [`ensure_https`] only validates the *initial* request URL; a `3xx` response can still
/// redirect the actual connection anywhere, including down to plain HTTP — or, per spec
/// `.local/specs/023-cargo-custom-registries/spec.md` NFR-003/plan-1b §1.1, straight to a
/// cloud metadata endpoint or other host no legitimate registry redirect ever targets. This
/// policy stops the redirect chain (rather than erroring) the moment a hop would do either,
/// so the caller sees the last successful `3xx` response and handles it exactly like any
/// other non-2xx status (`DepsError::HttpStatus`) instead of needing a distinct
/// "redirect blocked" error variant.
///
/// The blocked-host check (`hop_targets_blocked_host`) is unconditional and
/// policy-independent — it does not consult `guard` at all, since
/// [`HostClass::never_a_registry`](crate::net_policy::HostClass::never_a_registry)
/// is deliberately narrower than any workspace-registry policy setting: it blocks only the
/// classes (loopback, link-local, cloud metadata, unspecified) that are never a legitimate
/// registry redirect target for *any* ecosystem, benefiting every one of the eleven crates
/// sharing the baseline client, not only Cargo's workspace-declared indexes.
///
/// `guard`'s own [`AddrGuard::tier_allows`] term additionally rejects a hop whose target class
/// the *tier* does not allow — under [`AddrGuard::Baseline`] this term is constant `false`, so
/// every non-Cargo ecosystem (and Cargo's own `$CARGO_HOME`-provenance traffic) is
/// bit-for-bit unaffected; under [`AddrGuard::WorkspaceDeclared`] it closes the redirect-hop
/// half of issue #455 (an IP-literal hop to an RFC1918/CGNAT address, which `hyper-util`
/// parses directly and never routes through a resolver).
///
/// This only classifies the redirect target's URL string, not its DNS-resolved address — that
/// residual gap (issue #449, "D1" in PR #447's plan) is closed for the name-hop case by
/// [`BlockedAddrResolver`], which [`build_guarded_client`] wires into every client this module
/// builds: a redirect hop reuses the same `Client`, so its target's resolved address is
/// validated too, for free (FR-007).
///
/// Every other redirect — including cross-host ones, which are out of scope for this
/// policy — falls through to reqwest's default (`Policy::limited(10)`), preserving the
/// existing hop-count limit and mockito's plain-`http://` loopback chains
/// used throughout this module's tests.
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)
        }
    })
}

/// Redirect policy for a [`HttpCache::transport_for_origin`]-scoped client.
///
/// Stops any hop unless [`crate::net_policy::is_trusted_prefix`] accepts it against
/// `trusted_origin`, parsed once at [`Transport`] construction — origin equality (scheme,
/// host, and port must match exactly) **and** the hop's path lying at or under
/// `trusted_origin`'s own path, at a proper path-segment boundary.
///
/// Origin equality is the issue #795 fix: the pre-#795 behavior
/// (`attempt.url().as_str().starts_with(&trusted_origin)`, a raw string-prefix test) is
/// satisfied by `https://gitlab.mycorp.dev.evil.com/...`, `https://gitlab.mycorp.dev-evil.com/...`,
/// and `https://gitlab.mycorp.dev@evil.com/...` alike, even though only the last of those
/// three actually shares a host with `evil.com` — `Url::origin()` ignores userinfo and
/// matches scheme+host+port exactly, closing all three shapes at once. This also still
/// covers a downgrade to plain `http://`: an `http://` hop's origin can never equal an
/// `https://`-scheme trusted origin, which every current caller passes — a separate scheme
/// check (as [`redirect_policy`] has, for its no-trusted-origin case) would be dead code
/// here.
///
/// The path-segment-boundary check is **not** subsumed by origin equality, and deliberately
/// kept: a caller (NuGet's registration-hive/flat-container paging, or `deps-cargo`'s sparse
/// index, whose `RegistryIndex::as_str()` carries no trailing-slash guarantee either way —
/// issue #795 S1) pins to a specific *path* on a registry host that also serves other,
/// less-trusted paths, not merely to the host itself. A plain `str::starts_with` on the raw
/// path (this function's pre-S1-fix shape) is itself vulnerable to the same class of bug one
/// level down: a trusted path of `/cargo/index` would wrongly accept the same-origin sibling
/// `/cargo/index-public/steal` or `/cargo/indexEVIL`, since both start with the trusted
/// string textually. [`crate::net_policy::is_trusted_prefix`] requires a hop's path to equal
/// the trusted path or continue immediately after a `/` following it, closing that
/// regardless of whether `trusted_origin`'s own path happens to end in `/`.
///
/// A `trusted_origin` that fails to parse matches no hop — every redirect is stopped
/// (fail-closed) rather than treated as "no restriction" — logged once at construction time
/// via `tracing::warn!` rather than silently, since a caller-side bug producing an
/// unparseable `trusted_origin` would otherwise present only as every redirect on that
/// transport mysteriously failing. This is reachable today: `deps-nuget`'s Public tier builds
/// `trusted_prefix` from a service-index `@id` string that is never `Url`-validated (that
/// validation only runs for [`crate::net_policy::validate_index_url`]'s
/// `NuGetRegistryTier::WorkspaceDeclared` path), so a malformed `@id` reaches this parse.
fn trusted_origin_redirect_policy(trusted_origin: &str) -> reqwest::redirect::Policy {
    let trusted = match Url::parse(trusted_origin) {
        Ok(url) => Some(url),
        Err(error) => {
            tracing::warn!(
                trusted_origin = crate::net_policy::url_for_tracing(trusted_origin),
                %error,
                "trusted_origin failed to parse; every redirect hop on this transport will be rejected"
            );
            None
        }
    };
    reqwest::redirect::Policy::custom(move |attempt| {
        if is_trusted_origin(attempt.url(), trusted.as_ref()) {
            reqwest::redirect::Policy::default().redirect(attempt)
        } else {
            attempt.stop()
        }
    })
}

/// The origin-and-path decision [`trusted_origin_redirect_policy`]'s closure makes on every
/// redirect hop, extracted as a pure function so the bypass shapes from issue #795 can be
/// regression-tested directly against it — `reqwest::redirect::Attempt`'s fields are private
/// outside the `reqwest` crate, so the closure itself cannot be unit-tested without going
/// through a real HTTP round trip. Delegates to [`crate::net_policy::is_trusted_prefix`],
/// shared with `deps-nuget`'s registration-hive page `@id` pre-check (issue #795 S2).
fn is_trusted_origin(hop_url: &Url, trusted: Option<&Url>) -> bool {
    trusted.is_some_and(|t| crate::net_policy::is_trusted_prefix(hop_url, t))
}

/// Error returned by [`BlockedAddrResolver`] when a DNS resolution cannot be trusted for
/// connection use — either it produced no address, or at least one resolved address falls into
/// a blocked [`crate::net_policy::HostClass`].
///
/// Kept distinct from [`DepsError`] since this crosses into `reqwest::dns::Resolve`'s own
/// `BoxError` (`Box<dyn std::error::Error + Send + Sync>`) return type, not this crate's own
/// error type.
#[derive(Debug, thiserror::Error)]
enum ResolveGuardError {
    /// The resolver returned zero addresses for `host` — fail-closed (NFR-004) rather than
    /// silently treating "nothing resolved" as "nothing to block".
    #[error("DNS resolution for {host} returned no addresses")]
    NoAddresses { host: String },
    /// `addr`, resolved for `host`, falls into `class`, one of the
    /// [`crate::net_policy::HostClass::never_a_registry`] classes no legitimate registry index
    /// (or a redirect from one) could ever target.
    #[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,
    },
}

/// Validates every address `tokio::net::lookup_host` returned for `host`, rejecting the whole
/// resolution if any is blocked — an attacker's public A record alongside a blocked one must not
/// keep the probe alive (FR-003). `guard`'s tier additionally rejects a resolved address whose
/// class the *tier* does not allow (issue #455: an RFC1918/CGNAT-range name rebound at
/// connect time), on top of the policy-independent [`HostClass::never_a_registry`](crate::net_policy::HostClass::never_a_registry)
/// check every tier enforces.
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)
}

/// The synthetic-lookup function signature [`TestLookup`] wraps.
#[cfg(test)]
type SyntheticLookupFn = dyn Fn(&str) -> Vec<std::net::SocketAddr> + Send + Sync;

/// Test-only override for [`BlockedAddrResolver::resolve`], replacing `tokio::net::lookup_host`
/// with a synthetic lookup — lets a test exercise the resolver-guard wiring against an address
/// chosen by the test (e.g. an RFC1918 literal) without depending on real DNS. A newtype rather
/// than a hand-written `Debug` directly on [`BlockedAddrResolver`], so that struct's own
/// `#[derive(Debug)]` stays valid under both `cfg(test)` and not.
#[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(..)")
    }
}

/// Connect-time DNS resolver that closes the rebinding TOCTOU gap left by [`ensure_https`]/
/// [`hop_targets_blocked_host`]'s URL-string-only classification (issue #449): those check the
/// declared hostname, but `reqwest`'s connector resolves DNS independently, later, and an
/// attacker who controls the hostname's DNS can rebind it to a blocked address in between.
///
/// Wired into every client [`build_guarded_client`] returns, so all 11 ecosystem crates sharing
/// the baseline client pool inherit it with zero per-crate plumbing (FR-006/NFR-002).
///
/// # Scope
///
/// This resolver's guard tier decides how much a resolved address is scrutinized: under
/// [`AddrGuard::Baseline`] this enforces only the policy-independent
/// [`crate::net_policy::HostClass::never_a_registry`] tier (loopback, link-local,
/// cloud-metadata, unspecified), the same tier [`hop_targets_blocked_host`] already applies —
/// closing issue #449's filed exploit (cloud-metadata rebinding) but not full `PublicOnly`
/// semantics. Under [`AddrGuard::WorkspaceDeclared`], [`validate_resolved_addrs`] additionally
/// rejects any resolved address outside the snapshotted [`crate::net_policy::WorkspaceRegistryAccess`]
/// policy's allowed classes — closing issue #455 (a workspace-declared name that legitimately
/// resolves to `HostClass::Global` at parse time, then rebinds to an RFC1918/CGNAT address at
/// connect time).
///
/// # Fail-closed (NFR-004)
///
/// Returns `Err` — never `Ok`, never a fallback resolver — on a `lookup_host` error, zero
/// addresses, or any resolved address [`validate_resolved_addrs`] rejects for `self.guard`'s
/// tier.
///
/// # Known limitations
///
/// - [`ClientBuilder::resolve`](reqwest::ClientBuilder::resolve)/
///   [`resolve_to_addrs`](reqwest::ClientBuilder::resolve_to_addrs) overrides wrap *outside* the
///   configured resolver (`reqwest`'s `DnsResolverWithOverrides`) and would bypass this guard
///   entirely if ever called — this workspace does not call them today.
/// - A configured system proxy (`HTTPS_PROXY`) resolves the target hostname itself; this
///   resolver then only ever sees the proxy's own address. Operator configuration, not
///   attacker-controlled, so NOT claimed as a defended case.
/// - Never applies to an IP-literal host (`https://169.254.169.254/`): `hyper-util`'s connector
///   parses those directly and never calls the configured resolver, so
///   [`classify_host`](crate::net_policy::classify_host) (via [`ensure_https`]/
///   [`hop_targets_blocked_host`]/[`redirect_policy`]'s tier term) remains the sole guard for
///   literals — a disjoint domain from this resolver's name-based one, not a gap. This is also
///   why the cache layer gives [`HttpCache::get_cached_workspace`] zero protection against an
///   *initial* request URL that is itself an IP literal — see that method's docs.
/// - Unlike [`ensure_https`]/[`hop_targets_blocked_host`], this resolver has **no** `test-util`
///   carve-out for `Loopback`: it blocks a `localhost`/`127.0.0.1` *name* unconditionally, in
///   every build. A downstream `test-util` consumer that mocks by binding an IP literal (as
///   this workspace's own `mockito` usage does) is unaffected — literals never reach this
///   resolver at all — but one that mocks via a `localhost` *name* would be newly blocked.
#[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)
        })
    }
}

/// Builds a client with `HttpCache`'s shared configuration (user agent, timeout), varying the
/// redirect policy and resolver — kept in one place so a future client-wide setting (proxy,
/// connection pool sizing, etc.) can't silently miss any [`Transport`] this module builds. This
/// is also the workspace's only `Client::builder()` call site.
fn build_client_inner(
    redirect: reqwest::redirect::Policy,
    resolver: BlockedAddrResolver,
) -> Client {
    #[expect(
        clippy::expect_used,
        reason = "fixed, hardcoded client configuration — no attacker-influenced input; can \
                  only fail on a genuinely broken TLS backend, which is unrecoverable anyway"
    )]
    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")
}

/// Pairs one [`AddrGuard`] value with both halves it governs — its redirect policy and its
/// resolver — so a client whose redirect policy and resolver enforce different tiers cannot be
/// built by this function.
fn build_guarded_client(guard: AddrGuard) -> Client {
    build_client_inner(redirect_policy(guard), BlockedAddrResolver::new(guard))
}

/// Test-only variant of [`build_guarded_client`] that substitutes a synthetic DNS lookup for
/// `tokio::net::lookup_host` — shares [`build_client_inner`] with the production constructor, so
/// deleting the `.dns_resolver(...)` wiring from that shared function fails any test built on
/// this too, not just the production path.
#[cfg(test)]
fn build_guarded_client_with_lookup(guard: AddrGuard, lookup: TestLookup) -> Client {
    build_client_inner(
        redirect_policy(guard),
        BlockedAddrResolver::with_lookup(guard, lookup),
    )
}

/// A `Client` welded to the [`CacheTier`] its guard enforces.
///
/// [`Self::baseline`], [`Self::workspace`] and [`Self::origin_pinned`] are the only sanctioned
/// way to build one: each derives its redirect policy, its resolver and its tier from a single
/// [`AddrGuard`] value, so a mismatched pairing (e.g. a baseline-guarded client keyed under the
/// workspace cache namespace) never arises through normal construction — though `cache.rs` is
/// one module, so a hand-written `Transport { .. }` literal elsewhere in this file could still
/// mismatch them; the three constructors are what make that a deliberate act, not an accident
/// reachable by passing the wrong argument to an existing function.
#[derive(Clone)]
struct Transport {
    client: Client,
    tier: CacheTier,
}

impl Transport {
    /// The shared, unauthenticated transport used by every non-workspace request.
    fn baseline() -> Self {
        Self {
            client: build_guarded_client(AddrGuard::Baseline),
            tier: CacheTier::Baseline,
        }
    }

    /// The transport for Cargo's workspace-declared-registry requests, snapshotting `policy`'s
    /// current value once and sharing that single snapshot between the guard and the cache-key
    /// tier — see [`AddrGuard::WorkspaceDeclared`]'s docs for why this is a value snapshot, not
    /// a live `Arc` read, and why the guard and the tier must never snapshot independently.
    fn workspace(policy: &Arc<RegistryAccessPolicy>) -> Self {
        let snapshot = policy.get();
        Self {
            client: build_guarded_client(AddrGuard::WorkspaceDeclared(snapshot)),
            tier: CacheTier::WorkspaceDeclared(snapshot),
        }
    }

    /// The transport for one [`HttpCache::transport_for_origin`]-pinned origin: a plain
    /// [`AddrGuard::Baseline`] resolver, paired with [`trusted_origin_redirect_policy`] instead
    /// of [`redirect_policy`] — that policy pins by URL prefix, which already subsumes the
    /// blocked-host hop check, so this is the one documented caller of [`build_client_inner`]
    /// directly rather than [`build_guarded_client`].
    fn origin_pinned(trusted_origin: &str) -> Self {
        Self {
            client: build_client_inner(
                trusted_origin_redirect_policy(trusted_origin),
                BlockedAddrResolver::new(AddrGuard::Baseline),
            ),
            tier: CacheTier::Baseline,
        }
    }

    /// The transport for one origin-pinned **workspace-declared** host (issue #561/#562),
    /// optionally carrying a credential. Pairs [`trusted_origin_redirect_policy`] (send-scope
    /// confinement — no redirect hop may leave `trusted_origin`) with
    /// [`AddrGuard::WorkspaceDeclared`] (the connect-address policy guard, #455-class
    /// protection) and the namespaced [`CacheTier::Pinned`] tier. One constructor serves both
    /// #562's unauthenticated workspace-declared fetches (`authenticated: false`) and #561's
    /// authenticated ones (`authenticated: true`) — `authenticated` only affects
    /// [`HttpCache`]'s revalidation-eviction rule (FR-015), never `AddrGuard`/redirect
    /// confinement. The shipped [`Self::origin_pinned`] (public `api.nuget.org` path) is
    /// unaffected — this is a distinct constructor, not a modification of that one.
    fn origin_pinned_guarded(
        trusted_origin: &str,
        policy: &Arc<RegistryAccessPolicy>,
        authenticated: bool,
    ) -> Self {
        let snapshot = policy.get();
        Self {
            client: build_client_inner(
                trusted_origin_redirect_policy(trusted_origin),
                BlockedAddrResolver::new(AddrGuard::WorkspaceDeclared(snapshot)),
            ),
            tier: CacheTier::Pinned {
                digest: pinned_digest(trusted_origin, snapshot),
                authenticated,
            },
        }
    }
}

/// Identifies a `(trusted_origin, policy_snapshot)` pair for [`CacheTier::Pinned`] — **never**
/// a credential identity (see that variant's docs). Not cryptographically salted: unlike a
/// caller's own credential-header digest (e.g. `deps_nuget`'s salted `auth_id`), this value is
/// not attacker-observable secret material, only a pool/cache-key discriminant.
fn pinned_digest(trusted_origin: &str, snapshot: WorkspaceRegistryAccess) -> u64 {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    trusted_origin.hash(&mut hasher);
    snapshot.hash(&mut hasher);
    hasher.finish()
}

/// Reads a response body incrementally, aborting once it exceeds `limit`.
///
/// Chunked reading (via [`Response::chunk`]) is required because the
/// decompressed body size is not known upfront: `gzip` decoding strips
/// `Content-Length`, so the only reliable guard against an oversized or
/// maliciously amplified (decompression-bomb) response is counting bytes
/// as they arrive and bailing before the whole body is buffered.
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: RedactedUrl::new(url),
            source: e.into(),
        })?
    {
        if body.len() + chunk.len() > limit {
            return Err(DepsError::ResponseTooLarge {
                url: RedactedUrl::new(url),
                limit,
            });
        }
        body.extend_from_slice(&chunk);
    }

    Ok(body.freeze())
}

/// Cached HTTP response with validation headers.
///
/// Stores response body and cache validation headers (ETag, Last-Modified)
/// for efficient conditional requests. The body uses `Bytes` which is an
/// Arc-like type optimized for network data, enabling zero-cost cloning
/// across multiple consumers without copying.
///
/// # Examples
///
/// ```
/// use deps_core::cache::CachedResponse;
/// use bytes::Bytes;
/// use std::time::Instant;
///
/// let response = CachedResponse::new(Bytes::from("response data")).with_etag("\"abc123\"");
///
/// // Clone is cheap - only increments reference count
/// let cloned = response.clone();
/// ```
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct CachedResponse {
    /// Raw response body, shareable across consumers without copying.
    pub body: Bytes,
    /// `ETag` header from the response, used for `If-None-Match` revalidation.
    pub etag: Option<String>,
    /// `Last-Modified` header from the response, used for `If-Modified-Since` revalidation.
    pub last_modified: Option<String>,
    /// Local time the response was fetched, used for TTL expiry checks.
    pub fetched_at: Instant,
}

impl CachedResponse {
    /// Constructs a `CachedResponse` for `body`, fetched now, with [`Self::etag`] and
    /// [`Self::last_modified`] left `None` — chain [`Self::with_etag`] and/or
    /// [`Self::with_last_modified`] to attach them.
    ///
    /// Needed because [`Self`] is `#[non_exhaustive]`: a struct literal only works inside
    /// this crate, so every other crate must go through this constructor instead.
    ///
    /// # Examples
    ///
    /// ```
    /// use deps_core::cache::CachedResponse;
    /// use bytes::Bytes;
    ///
    /// let response = CachedResponse::new(Bytes::from("response data")).with_etag("\"abc123\"");
    /// assert_eq!(response.etag.as_deref(), Some("\"abc123\""));
    /// ```
    #[must_use]
    pub fn new(body: Bytes) -> Self {
        Self {
            body,
            etag: None,
            last_modified: None,
            fetched_at: Instant::now(),
        }
    }

    /// Attaches the response's `ETag` header. See [`Self::etag`].
    #[must_use]
    pub fn with_etag(mut self, etag: impl Into<String>) -> Self {
        self.etag = Some(etag.into());
        self
    }

    /// Attaches the response's `Last-Modified` header. See [`Self::last_modified`].
    #[must_use]
    pub fn with_last_modified(mut self, last_modified: impl Into<String>) -> Self {
        self.last_modified = Some(last_modified.into());
        self
    }

    /// Overrides when the response was fetched — otherwise [`Self::new`] stamps
    /// [`Instant::now`]. Lets external code (tests, benches) construct a backdated entry to
    /// exercise TTL-expiry behavior, the one thing a direct struct literal used to allow.
    #[must_use]
    pub const fn with_fetched_at(mut self, fetched_at: Instant) -> Self {
        self.fetched_at = fetched_at;
        self
    }
}

/// HTTP cache with ETag and Last-Modified validation.
///
/// Implements RFC 7232 conditional requests to minimize network traffic.
/// All responses are cached with their validation headers, and subsequent
/// requests use `If-None-Match` (ETag) or `If-Modified-Since` headers
/// to check for updates.
///
/// The cache uses `Bytes` for response bodies, enabling efficient sharing
/// of cached data across multiple consumers without copying. `Bytes` is
/// an Arc-like type optimized for network I/O.
///
/// # Examples
///
/// ```no_run
/// use deps_core::cache::HttpCache;
///
/// # async fn example() -> deps_core::error::Result<()> {
/// let cache = HttpCache::new();
///
/// // First request - fetches from network
/// let data1 = cache.get_cached("https://index.crates.io/se/rd/serde").await?;
///
/// // Second request - uses conditional GET (304 Not Modified if unchanged)
/// let data2 = cache.get_cached("https://index.crates.io/se/rd/serde").await?;
/// # Ok(())
/// # }
/// ```
///
/// # Cache key
///
/// Entries are keyed by URL alone (see `Self::cache_key`, private) — `extra_headers` (see
/// [`HttpCache::get_cached_with_headers`]) play no part in the cache key.
/// This is safe only as long as "same URL" implies "same representation":
/// a content-negotiating header (e.g. a per-request `Accept`) that can vary
/// the response body for an otherwise-identical URL requires giving each
/// distinct representation its own URL (e.g. a query parameter or distinct
/// path), not just a distinct header value, or callers requesting different
/// representations of the same URL will silently share one cache entry.
///
/// Likewise, the key doesn't encode *which* client (and so which redirect policy)
/// produced an entry — [`HttpCache::get_cached`] and [`HttpCache::get_cached_trusted_origin`]
/// share one entry map. No caller today requests the same URL through both, but one that did
/// could observe the other's cached (and differently redirect-validated) body.
///
/// [`Self::get_cached_workspace`] is the one exception: it is namespaced under a distinct,
/// policy-scoped key prefix (see `Self::cache_key`, private) so a body fetched under a looser
/// [`crate::net_policy::WorkspaceRegistryAccess`] can never be served back once the policy
/// tightens.
pub struct HttpCache {
    entries: DashMap<String, CachedResponse>,
    /// Running total of `body.len()` across all `entries`, kept in sync by
    /// [`HttpCache::store_entry`], [`HttpCache::evict_entries`], and
    /// [`HttpCache::clear`] via relative `fetch_add`/`fetch_sub` only —
    /// never an absolute `store` after the initial `0`, since that would
    /// silently discard any concurrent relative update racing with it. Used
    /// to trigger byte-bounded eviction without summing every entry on each
    /// check; advisory (see [`MAX_CACHE_BYTES`]), not an exact live count
    /// under concurrent access.
    total_bytes: AtomicUsize,
    /// The shared, unauthenticated transport used by every non-workspace request.
    baseline: Transport,
    /// Per-`(trusted_origin, tier)` transport pool backing [`Self::get_cached_trusted_origin`]
    /// and [`Self::get_cached_pinned`] alike (issue #561/#562, FR-017), keyed by the exact
    /// `trusted_origin` prefix string passed to that call, paired with the [`CacheTier`] it was
    /// built for. reqwest's redirect policy is fixed per-`Client`, so a distinct client is
    /// unavoidable per distinct origin; pooled here so repeated calls against the same
    /// `(origin, tier)` reuse one transport (and its connection pool) instead of rebuilding on
    /// every call. Deliberately **uncapped** — see [`Self::set_registry_policy`]'s docs for why
    /// a capacity cap was considered and dropped for this pool.
    trusted_clients: DashMap<(String, CacheTier), Transport>,
    /// Live-updatable Cargo workspace-registry policy the `workspace` transport field below
    /// and cache-key namespace are derived from. Kept alongside that field (not just read once
    /// at construction) so [`Self::cache_key`] and [`Self::set_registry_policy`] both read the
    /// same discriminant.
    policy: Arc<RegistryAccessPolicy>,
    /// The transport for [`Self::get_cached_workspace`], rebuilt in place by
    /// [`Self::set_registry_policy`] on every actual policy transition.
    workspace: RwLock<Transport>,
    /// Test-only counter of how many times [`Self::set_registry_policy`] has actually rebuilt
    /// the `workspace` transport field above (as opposed to no-op'ing on an unchanged value) —
    /// asserts C4's rebuild-only-on-change behavior directly.
    #[cfg(test)]
    workspace_rebuilds: AtomicUsize,
    /// Live-updatable "no outbound requests" flag (issue #483). Enforced by
    /// [`Self::ensure_online`] at all 4 send sites. See [`Self::set_offline`]'s docs for
    /// the override this has on `cache_enabled` below.
    offline: AtomicBool,
    /// Live-updatable entry-map toggle (issue #482): `false` bypasses the entry map
    /// entirely (see `get_cached_with_headers_via`). Overridden to effectively `true`
    /// whenever `offline` is set — see [`Self::set_offline`]'s docs.
    cache_enabled: AtomicBool,
}

impl HttpCache {
    /// Creates a new HTTP cache with default configuration and the default
    /// [`crate::net_policy::WorkspaceRegistryAccess`] policy (`PublicOnly`).
    ///
    /// The cache uses a configurable timeout for all requests and identifies
    /// itself with an auto-versioned user agent.
    pub fn new() -> Self {
        Self::with_policy(Arc::new(RegistryAccessPolicy::default()))
    }

    /// Creates a new HTTP cache whose [`Self::get_cached_workspace`] requests are governed by
    /// `policy`'s live value.
    ///
    /// A later [`Self::set_registry_policy`] call rebuilds the workspace transport (and its
    /// cache-key namespace) in place, so every caller holding this `HttpCache` sees the new
    /// policy take effect immediately, with no need to reconstruct the cache.
    ///
    /// # Examples
    ///
    /// ```
    /// use deps_core::HttpCache;
    /// use deps_core::net_policy::RegistryAccessPolicy;
    /// use std::sync::Arc;
    ///
    /// let policy = Arc::new(RegistryAccessPolicy::default());
    /// let cache = HttpCache::with_policy(Arc::clone(&policy));
    /// assert!(cache.is_empty());
    /// ```
    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),
        }
    }

    /// Sets whether outbound network requests are permitted (issue #483).
    ///
    /// Enforced by `Self::ensure_online` (private) at every one of this module's 4 send sites —
    /// effective for every call after this returns. While `value` is `true`, this also
    /// overrides `cache_enabled` (see [`Self::set_cache_enabled`]) to behave as `true` on
    /// both the read and write path in `get_cached_with_headers_via`: without this, a
    /// warm entry fetched before going offline could never have been stored in the first
    /// place if caching was disabled, leaving the offline warm-cache path with nothing to
    /// serve — the exact combination `cache.enabled: false` + `network.offline: true` is
    /// meant to survive.
    pub fn set_offline(&self, value: bool) {
        self.offline.store(value, Ordering::Relaxed);
    }

    /// Returns whether outbound network requests are currently blocked.
    #[must_use]
    pub fn is_offline(&self) -> bool {
        self.offline.load(Ordering::Relaxed)
    }

    /// Sets whether the entry-map cache is used (issue #482). See [`Self::set_offline`]'s
    /// docs for the override `offline` has on this flag while set.
    pub fn set_cache_enabled(&self, value: bool) {
        self.cache_enabled.store(value, Ordering::Relaxed);
    }

    /// Returns `Err(DepsError::Offline)` when `network.offline` is set, without making any
    /// request — the last check before a socket opens at each of this module's 4 send
    /// sites, placed beside the existing [`ensure_https`] call at each.
    fn ensure_online(&self, url: &str) -> Result<()> {
        if self.is_offline() {
            return Err(DepsError::Offline {
                url: RedactedUrl::new(url),
            });
        }
        Ok(())
    }

    /// Returns the transport scoped to `trusted_origin`, building and pooling one on first use.
    ///
    /// The `get` fast path (a shared read lock) serves the common case — a `trusted_origin`
    /// already pooled — without ever taking `trusted_clients`' write-capable `entry` lock;
    /// `entry().or_insert_with()` only runs on a miss, so two callers racing on the same new
    /// origin still only ever build and store one [`Transport`] for it, not one each.
    fn transport_for_origin(&self, trusted_origin: &str) -> Transport {
        let key = (trusted_origin.to_string(), CacheTier::Baseline);
        if let Some(existing) = self.trusted_clients.get(&key) {
            return existing.clone();
        }

        self.trusted_clients
            .entry(key)
            .or_insert_with(|| Transport::origin_pinned(trusted_origin))
            .clone()
    }

    /// Like [`Self::transport_for_origin`], but for an origin-pinned, connect-address-guarded
    /// [`CacheTier::Pinned`] transport (issue #561/#562) — building and pooling one on first
    /// use, keyed by `(trusted_origin, CacheTier::Pinned { .. })` so an authenticated and
    /// unauthenticated transport for the same origin are pooled separately.
    fn transport_for_pinned(&self, trusted_origin: &str, authenticated: bool) -> Transport {
        let digest = pinned_digest(trusted_origin, self.policy.get());
        let key = (
            trusted_origin.to_string(),
            CacheTier::Pinned {
                digest,
                authenticated,
            },
        );
        if let Some(existing) = self.trusted_clients.get(&key) {
            return existing.clone();
        }

        self.trusted_clients
            .entry(key)
            .or_insert_with(|| {
                Transport::origin_pinned_guarded(trusted_origin, &self.policy, authenticated)
            })
            .clone()
    }

    /// The prefix marking a workspace-tier cache key, chosen as a control character that can
    /// never appear at the start of a URL string this module writes: production code paths
    /// only ever write keys derived from this function, which are either the bare URL (starting
    /// `https://`, or — test cfgs only — `http://` on loopback) or this prefix followed by a
    /// policy digit. No in-process caller other than [`Self::insert_for_bench`] (a
    /// `#[doc(hidden)]` test/bench helper that accepts a caller-chosen key) can write an
    /// arbitrary key, so a `Baseline`-tier and `WorkspaceDeclared`-tier entry can never collide
    /// in production use.
    const WS_KEY_PREFIX: char = '\u{1}';

    /// The prefix marking a [`CacheTier::Pinned`]-tier cache key (issue #561/#562) — distinct
    /// from [`Self::WS_KEY_PREFIX`] so the two namespaces can never collide, chosen as another
    /// control character no URL string this module writes can start with.
    const PINNED_KEY_PREFIX: char = '\u{2}';

    /// Computes the cache-map key for `url` under `tier` — `Cow::Borrowed(url)` for
    /// [`CacheTier::Baseline`] (allocation-free, and identical to every entry this cache wrote
    /// before this policy-tier split existed), or a policy-digit-prefixed owned key for
    /// [`CacheTier::WorkspaceDeclared`] so a policy tightening can never serve a body fetched
    /// under a looser policy (C5): the digit comes from `tier`'s own snapshot — the exact same
    /// value the paired [`Transport`]'s [`AddrGuard`] enforced for this request, taken together
    /// at [`Transport::workspace`] construction time — never a separate live `self.policy` read,
    /// which would open a read-skew window between the guard that let a fetch through and the
    /// key that fetch's body gets stored under. `self.policy` remains this cache's live handle
    /// for [`Self::set_registry_policy`]'s change detection and the write-through source that
    /// method snapshots from when rebuilding the workspace transport — never consulted here.
    ///
    /// Callers must compute this once per request and thread the result through, never
    /// recompute mid-request — a policy flip between two recomputations would read and write
    /// under different keys for what should be one atomic operation.
    ///
    /// `auth_id` (FR-014) is folded in only for [`CacheTier::Pinned`] — a separate credential
    /// identity from `digest` (see that variant's docs), so a rotated or distinct credential
    /// against the same origin never reads back a body fetched under a different one.
    /// `authenticated` (the tier's own field) is folded in too, so an authenticated and an
    /// unauthenticated fetch against the same origin/`auth_id` can never share an entry either
    /// — relevant only in principle today (every current caller correlates `authenticated` and
    /// `auth_id.is_some()`), but [`HttpCache::get_cached_pinned_with_headers`] is a public API
    /// with no enforced invariant tying the two together, so the key format does not assume one.
    ///
    /// Both `auth_id` and `authenticated` are encoded as one-char tags immediately preceding a
    /// fixed-width field (`0`/`1` before a 16-hex-digit `auth_id` field, `U`/`A` right before
    /// `url`), rather than collapsing `auth_id: None` to a `0` sentinel value:
    /// [`auth_digest`](crate::secret::auth_digest) is a non-cryptographic hash, so `Some(0)` is
    /// a legitimate (if astronomically unlikely) digest a real credential can produce, and
    /// `unwrap_or(0)` would make that fetch's cache key identical to an unauthenticated one —
    /// letting a cached authenticated body be served to an unauthenticated request, or vice
    /// versa (issue #1025). Every hex field is written at a fixed width (`digest`, `auth_id`),
    /// so no tag or field can be confused with part of `digest`, the `auth_id` field, or `url`
    /// — a variable-width field would let two distinct `(digest, auth_id)` pairs produce the
    /// same concatenated string.
    fn cache_key<'a>(&self, url: &'a str, tier: CacheTier, auth_id: Option<u64>) -> Cow<'a, str> {
        match tier {
            CacheTier::Baseline => Cow::Borrowed(url),
            CacheTier::WorkspaceDeclared(snapshot) => {
                Cow::Owned(format!("{}{}{url}", Self::WS_KEY_PREFIX, snapshot.to_u8()))
            }
            CacheTier::Pinned {
                digest,
                authenticated,
            } => {
                let (auth_tag, id) = match auth_id {
                    Some(id) => ('1', id),
                    None => ('0', 0),
                };
                let authenticated_tag = if authenticated { 'A' } else { 'U' };
                Cow::Owned(format!(
                    "{}{digest:016x}{auth_tag}{id:016x}{authenticated_tag}{url}",
                    Self::PINNED_KEY_PREFIX,
                ))
            }
        }
    }

    /// Retrieves data from URL with intelligent caching.
    ///
    /// On first request, fetches data from the network and caches it.
    /// On subsequent requests, performs a conditional GET request using
    /// cached ETag or Last-Modified headers. If the server responds with
    /// 304 Not Modified, returns the cached data. Otherwise, fetches and
    /// caches the new data.
    ///
    /// If the conditional request fails due to network errors, falls back
    /// to the cached data (stale-while-revalidate pattern).
    ///
    /// # Returns
    ///
    /// Returns `Bytes` containing the response body. Multiple calls for the
    /// same URL return cheap clones (reference counting) without copying data.
    ///
    /// # Errors
    ///
    /// Returns `DepsError::RegistryError` if the initial fetch fails and no
    /// cached data exists, `DepsError::HttpStatus` if the server returns a
    /// non-2xx status on that initial fetch, or `DepsError::ResponseTooLarge`
    /// if the response body exceeds the configured size cap.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use deps_core::cache::HttpCache;
    /// # async fn example() -> deps_core::error::Result<()> {
    /// let cache = HttpCache::new();
    /// let data = cache.get_cached("https://example.com/api/data").await?;
    /// println!("Fetched {} bytes", data.len());
    /// # Ok(())
    /// # }
    /// ```
    pub async fn get_cached(&self, url: &str) -> Result<Bytes> {
        self.get_cached_with_headers(url, &[]).await
    }

    /// Returns the cached body for `url` without making any network request.
    ///
    /// Unlike `get_cached`'s own stale-while-revalidate fallback (the `Err` arm of
    /// `conditional_request_with_headers`'s match in `get_cached_with_headers_via`), this
    /// is reachable even when a caller wraps `get_cached` in a short outer timeout: a
    /// hung conditional request that never resolves within that timeout gets its whole
    /// future cancelled, so `get_cached`'s internal fallback logic never runs and the
    /// caller sees a timeout instead of stale data. A caller in that position can call
    /// this instead — a synchronous map lookup, no I/O — to serve the last known-good
    /// body itself. Returns `None` if `url` has never been successfully cached.
    ///
    /// The returned body carries no age bound: this bypasses `get_cached`'s own
    /// freshness/revalidation logic entirely, so a caller that surfaces this body to
    /// the user (e.g. inserting it into a manifest edit) should treat it as
    /// arbitrarily stale, not just-expired.
    ///
    /// Reads the baseline (unprefixed) cache-key namespace only (see `Self::cache_key`, private) — a
    /// body fetched via [`Self::get_cached_workspace`] is never visible through this method.
    #[must_use]
    pub fn peek_cached(&self, url: &str) -> Option<Bytes> {
        self.entries.get(url).map(|r| r.body.clone())
    }

    /// Fetches a URL with additional request headers, using the cache.
    ///
    /// Works the same as `get_cached` but injects extra headers (e.g., Authorization)
    /// into every request. Useful for APIs that require authentication tokens.
    ///
    /// # Errors
    ///
    /// Returns `DepsError::RegistryError` if the initial fetch fails and no
    /// cached data exists, `DepsError::HttpStatus` if the server returns a
    /// non-2xx status on that initial fetch, or `DepsError::ResponseTooLarge`
    /// if the response body exceeds the configured size cap.
    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, None)
            .await
    }

    /// Like [`Self::get_cached`], but additionally stops any redirect hop that does not match
    /// `trusted_origin` via [`crate::net_policy::is_trusted_prefix`] (origin equality plus a
    /// path-segment-boundary prefix check, e.g. against `https://api.nuget.org/v3/registration5-gz/`).
    ///
    /// For a caller that already validated the *initial* request URL against a trusted
    /// prefix (NuGet's registration-hive paging validates `page.id` this way) and needs
    /// that guarantee to hold through any redirect too, not just the first hop —
    /// [`Self::get_cached`]'s own policy deliberately does not enforce this, since
    /// cross-host redirects are legitimate for the other registry clients sharing this
    /// cache; the stricter check is opt-in per call rather than global.
    ///
    /// The block only surfaces as an error on a cold cache: like [`Self::get_cached`]'s own
    /// stale-while-revalidate fallback, a warm entry for `url` still returns the last
    /// known-good body (itself already fetched and origin-validated on a prior call)
    /// instead of propagating a blocked-redirect `HttpStatus` from a revalidation attempt.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_cached`].
    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, None)
            .await
    }

    /// Like [`Self::get_cached_trusted_origin`], but additionally injects `extra_headers`
    /// (e.g. an `Authorization` bearer token) into every request — the authenticated
    /// counterpart to [`Self::get_cached_with_headers`], composed with the same
    /// origin-pinned redirect policy [`Self::get_cached_trusted_origin`] uses.
    ///
    /// This exists specifically so a header carrying a credential can never survive a
    /// cross-origin redirect hop: [`Self::get_cached_with_headers`] attaches
    /// `extra_headers` to the *initial* request only and follows reqwest's default
    /// (same-scheme, any-host) redirect policy for every hop after that, which is
    /// exactly the shape a hostile or misconfigured redirect on the resolved index
    /// itself could exploit to exfiltrate a bearer token to an attacker-controlled
    /// host. Composing `Self::transport_for_origin`'s (private) pinned-origin transport with header
    /// injection closes that by construction — no empirical redirect test is needed
    /// to prove the header cannot leak, since the client stops following before a
    /// cross-origin hop would ever be sent.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_cached_trusted_origin`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use deps_core::cache::HttpCache;
    /// use reqwest::header;
    ///
    /// # async fn example() -> deps_core::error::Result<()> {
    /// let cache = HttpCache::new();
    /// let data = cache
    ///     .get_cached_trusted_origin_with_headers(
    ///         "https://index.mycorp.dev/se/rd/serde",
    ///         "https://index.mycorp.dev/",
    ///         &[(header::AUTHORIZATION, "Bearer secret-token")],
    ///     )
    ///     .await?;
    /// println!("Fetched {} bytes", data.len());
    /// # Ok(())
    /// # }
    /// ```
    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, None)
            .await
    }

    /// Like [`Self::get_cached_trusted_origin_with_headers`], but for an origin-pinned,
    /// connect-address-guarded `CacheTier::Pinned` transport (issue #561/#562) instead of the
    /// baseline-guarded [`Self::get_cached_trusted_origin`] one — the only sanctioned way to
    /// send a credential to a workspace-declared host. Delegates to
    /// [`Self::get_cached_pinned_with_headers`] with no extra headers.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_cached`].
    pub async fn get_cached_pinned(
        &self,
        url: &str,
        trusted_origin: &str,
        authenticated: bool,
        auth_id: Option<u64>,
    ) -> Result<Bytes> {
        self.get_cached_pinned_with_headers(url, trusted_origin, authenticated, auth_id, &[])
            .await
    }

    /// Like [`Self::get_cached_pinned`], but additionally injects `extra_headers` (e.g. an
    /// `Authorization` header carrying a credential) into every request — composed with the
    /// same origin-pinned, connect-address-guarded transport [`Self::get_cached_pinned`] uses,
    /// so a credential header can never survive a cross-origin redirect hop, exactly like
    /// [`Self::get_cached_trusted_origin_with_headers`]'s identical closure argument for the
    /// baseline-guarded tier.
    ///
    /// `auth_id` (FR-014) — a caller-computed, salted digest of the credential actually being
    /// attached (`None` for an unauthenticated #562 fetch) — is folded into the cache key only,
    /// never into `CacheTier`/the transport-pool key, so a rotated or distinct credential
    /// against the same origin never reads back a body fetched under a different one.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_cached`].
    pub async fn get_cached_pinned_with_headers(
        &self,
        url: &str,
        trusted_origin: &str,
        authenticated: bool,
        auth_id: Option<u64>,
        extra_headers: &[(header::HeaderName, &str)],
    ) -> Result<Bytes> {
        let transport = self.transport_for_pinned(trusted_origin, authenticated);
        self.get_cached_with_headers_via(url, extra_headers, &transport, auth_id)
            .await
    }

    /// Like [`Self::get_cached`], but for Cargo's workspace-declared-registry requests: routes
    /// through the workspace transport field, whose guard enforces the live
    /// [`crate::net_policy::WorkspaceRegistryAccess`] policy on both the resolved connect-time
    /// address (issue #455) and any redirect hop, and keys the entry under a
    /// policy-scoped namespace (see `Self::cache_key`, private) distinct from every other method on
    /// this cache.
    ///
    /// This gives the *resolved address* the same policy scrutiny `deps_cargo::config::RegistryIndex::new`
    /// already gives the declared URL string at parse time — it does **not**
    /// re-check the initial request URL itself: a caller passing an IP-literal `url` whose
    /// class the policy would reject connects anyway, since `hyper-util`'s connector parses an
    /// IP literal directly and never calls the configured resolver (see
    /// `BlockedAddrResolver`'s docs, private). `RegistryIndex::new` is the sole, by-design gate for
    /// that residual — every caller of this method already went through it.
    ///
    /// If an entry is already cached, a revalidation failure — including a guard rejection
    /// from a since-rebound or since-tightened-policy address — falls back to serving the
    /// cached body, logging only a `tracing::warn!` (pre-existing behavior, unrelated to
    /// this method). This is not a bypass — no new connection to the blocked address is
    /// made — but it means such a block is invisible to the caller whenever an entry already
    /// exists for that URL.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_cached`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use deps_core::HttpCache;
    /// use deps_core::net_policy::RegistryAccessPolicy;
    /// use std::sync::Arc;
    ///
    /// # async fn example() -> deps_core::error::Result<()> {
    /// let policy = Arc::new(RegistryAccessPolicy::default());
    /// let cache = HttpCache::with_policy(policy);
    /// let data = cache
    ///     .get_cached_workspace("https://index.mycorp.dev/se/rd/serde")
    ///     .await?;
    /// println!("Fetched {} bytes", data.len());
    /// # Ok(())
    /// # }
    /// ```
    pub async fn get_cached_workspace(&self, url: &str) -> Result<Bytes> {
        self.get_cached_workspace_with_headers(url, &[]).await
    }

    /// Like [`Self::get_cached_workspace`], but additionally forwards `extra_headers` to the
    /// underlying request — the headered form needed by a registry client whose
    /// workspace-declared fetch requires a non-default header (e.g. `deps-npm`'s abbreviated-
    /// packument `Accept` header for an alternate npm registry).
    ///
    /// # Security
    ///
    /// `extra_headers` are attached to the **initial** request only. The workspace transport
    /// pins by [`crate::net_policy::HostClass`], not origin — unlike
    /// [`Self::get_cached_trusted_origin_with_headers`], which exists precisely to close this
    /// gap for a caller that needs it — so a cross-origin redirect hop to any other
    /// policy-permitted host is followed with `extra_headers` re-sent by reqwest's default
    /// redirect policy. **This method must never carry a credential.** Harmless for its
    /// current sole caller (a fixed `Accept` header), but directly load-bearing for any
    /// future auth-wiring work: reach for [`Self::get_cached_trusted_origin_with_headers`]
    /// instead if a header ever needs to stay pinned to one origin.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_cached`].
    pub async fn get_cached_workspace_with_headers(
        &self,
        url: &str,
        extra_headers: &[(header::HeaderName, &str)],
    ) -> Result<Bytes> {
        #[expect(
            clippy::expect_used,
            reason = "poisoned only if another thread already panicked holding the lock — \
                      propagate via panic, matching RwLock's poisoning contract"
        )]
        let transport = self
            .workspace
            .read()
            .expect("workspace transport lock poisoned")
            .clone();
        self.get_cached_with_headers_via(url, extra_headers, &transport, None)
            .await
    }

    /// Updates the policy governing [`Self::get_cached_workspace`], rebuilding the workspace
    /// transport field (and so its cache-key namespace and guard together) when
    /// `value` actually differs from the current setting — a no-op call does not rebuild, so a
    /// caller that re-applies an unchanged configuration does not pay for a fresh `Client` and
    /// its connection pool.
    ///
    /// Effective for every [`Self::get_cached_workspace`] call after this returns. Note this
    /// only gates *future* fetches: an `All -> PublicOnly`/`Off` tightening does not purge
    /// already-registered `deps-cargo` alternate-registry clients resolved under the looser
    /// policy (pre-existing, documented on [`crate::net_policy::RegistryAccessPolicy::set`]).
    ///
    /// Unlike that pre-existing gap, every `CacheTier::Pinned` cache entry (issue #561/#562)
    /// **is** purged on every actual policy transition, along with every pinned-tier pooled
    /// `Transport` — substantially narrowing the `All -> PublicOnly -> All` round-trip hole for
    /// credential-carrying entries specifically (NFR-004): re-namespacing alone (as the
    /// pre-existing workspace-tier digit prefix does) would leave an old-era authenticated
    /// body reachable once the policy round-trips back to a value whose digest happens to
    /// collide again. Not an absolute close: a fetch already in flight when the transition
    /// happens can still land its response and re-insert an old-era key after the purge —
    /// harmless (readable only under the era it was legitimately fetched in), just not
    /// prevented by this purge alone.
    pub fn set_registry_policy(&self, value: WorkspaceRegistryAccess) {
        if self.policy.get() == value {
            return;
        }
        self.policy.set(value);
        let rebuilt = Transport::workspace(&self.policy);
        #[expect(
            clippy::expect_used,
            reason = "poisoned only if another thread panicked while holding the lock — see \
                      the matching justification on get_cached_workspace_with_headers above"
        )]
        {
            *self
                .workspace
                .write()
                .expect("workspace transport lock poisoned") = rebuilt;
        }
        #[cfg(test)]
        self.workspace_rebuilds.fetch_add(1, Ordering::Relaxed);

        let mut freed_bytes = 0usize;
        self.entries.retain(|k, v| {
            let keep = !k.starts_with(Self::PINNED_KEY_PREFIX);
            if !keep {
                freed_bytes += v.body.len();
            }
            keep
        });
        self.total_bytes.fetch_sub(freed_bytes, Ordering::Relaxed);
        self.trusted_clients
            .retain(|(_, tier), _| !matches!(tier, CacheTier::Pinned { .. }));
    }

    /// `auth_id` (FR-014) is meaningful only under [`CacheTier::Pinned`] — every other tier
    /// ignores it (see [`Self::cache_key`]'s docs).
    #[tracing::instrument(
        skip(self, extra_headers, transport, auth_id),
        fields(url = %RedactedUrl::new(url), cache = tracing::field::Empty)
    )]
    async fn get_cached_with_headers_via(
        &self,
        url: &str,
        extra_headers: &[(header::HeaderName, &str)],
        transport: &Transport,
        auth_id: Option<u64>,
    ) -> 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();
        // `offline` forces `cache_enabled` true (S1 fix): otherwise `cache.enabled: false`
        // would leave nothing to serve offline for a URL fetched while caching was disabled.
        // See `Self::set_offline`'s docs.
        let cache_enabled = self.cache_enabled.load(Ordering::Relaxed) || offline;

        // Computed once and threaded through every call — recomputing mid-request could read
        // and write under different keys (see `Self::cache_key`'s docs).
        let cache_key = self.cache_key(url, transport.tier, auth_id);

        if !cache_enabled {
            // Explicit, not `Empty`: an empty `cache` field would be indistinguishable from
            // broken instrumentation (#756 S3) on a path that's neither a hit nor a miss.
            tracing::Span::current().record("cache", "disabled");
            return self
                .transport_only_via(url, extra_headers, BodyLimit::DEFAULT, &transport.client)
                .await;
        }

        // Clone+drop the Ref immediately: holding it across `.await` can deadlock a concurrent
        // task needing write access to the same shard.
        if let Some(cached) = self.entries.get(cache_key.as_ref()).map(|r| r.clone()) {
            if offline {
                // Skips the conditional-request attempt: `ensure_online` would block it and
                // fall back to this same body anyway, just with a spurious warn + wasted
                // allocation. The only branch below that is a genuine zero-network "hit"
                // (#756 S3); every other branch still issued a request.
                tracing::Span::current().record("cache", "hit");
                return Ok(cached.body);
            }
            match self
                .conditional_request_with_headers(
                    url,
                    &cached,
                    extra_headers,
                    &transport.client,
                    &cache_key,
                )
                .await
            {
                // 304: a round trip happened but no body was re-transferred — distinct from
                // `hit` (no request) and `refreshed` (full re-fetch).
                Ok(None) => {
                    tracing::Span::current().record("cache", "revalidated");
                    return Ok(cached.body);
                }
                // Stale entry cost a full re-fetch, same as a miss — must not report as a hit.
                Ok(Some(new_body)) => {
                    tracing::Span::current().record("cache", "refreshed");
                    return Ok(new_body);
                }
                Err(e) => {
                    // FR-015/NFR-004: a 401/403 revalidation against an *authenticated*
                    // pinned-tier entry must evict rather than serve the possibly-revoked
                    // credential's last-known-good body — every other tier keeps today's
                    // stale-while-revalidate fallback unchanged.
                    if transport.tier.is_authenticated()
                        && matches!(
                            &e,
                            DepsError::HttpStatus {
                                status: 401 | 403,
                                ..
                            }
                        )
                    {
                        if let Some((_, old)) = self.entries.remove(cache_key.as_ref()) {
                            self.total_bytes
                                .fetch_sub(old.body.len(), Ordering::Relaxed);
                        }
                        tracing::Span::current().record("cache", "evicted");
                        // #756 round 2 S1: never interpolate `e` — its `Display`/`Debug` embed
                        // the raw unredacted `url`, defeating this span's `RedactedUrl` field.
                        // `safe_tracing_summary` extracts only the safe status+cause instead.
                        let (status, cause) = e.safe_tracing_summary();
                        tracing::warn!(
                            status = ?status,
                            cause,
                            "evicting authenticated cache entry after revalidation failure"
                        );
                        return Err(e);
                    }
                    tracing::Span::current().record("cache", "stale-fallback");
                    // Same rationale as above: no `e` interpolation.
                    let (status, cause) = e.safe_tracing_summary();
                    tracing::warn!(
                        status = ?status,
                        cause,
                        "conditional request failed, using cache"
                    );
                    return Ok(cached.body);
                }
            }
        }

        tracing::Span::current().record("cache", "miss");
        self.fetch_and_store_with_headers(url, extra_headers, &transport.client, &cache_key)
            .await
    }

    /// Performs conditional HTTP request using cached validation headers.
    ///
    /// Sends `If-None-Match` (ETag) and/or `If-Modified-Since` headers
    /// to check if the cached content is still valid.
    ///
    /// # Returns
    ///
    /// - `Ok(Some(Bytes))` - Server returned 200 OK with new content
    /// - `Ok(None)` - Server returned 304 Not Modified (cache is valid)
    /// - `Err(_)` - Network or HTTP error occurred
    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: RedactedUrl::new(url),
            source: e.into(),
        })?;

        if response.status() == StatusCode::NOT_MODIFIED {
            return Ok(None);
        }

        if !response.status().is_success() {
            return Err(DepsError::HttpStatus {
                url: RedactedUrl::new(url),
                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))
    }

    /// Fetches a fresh response from the network and stores it in the cache.
    ///
    /// This method bypasses the cache and always makes a network request.
    /// The response is stored with its ETag and Last-Modified headers for
    /// future conditional requests.
    ///
    /// # Errors
    ///
    /// Returns `DepsError::HttpStatus` if the server returns a non-2xx status code,
    /// `DepsError::RegistryError` if the network request fails, or
    /// `DepsError::ResponseTooLarge` if the response body exceeds the
    /// configured size cap.
    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)?;
        // #756 security follow-up (S-A): `RedactedUrl`, not the raw `url` — this is the
        // direct callee of the now-hardened `get_cached_with_headers_via`, and at `debug`
        // level, which is this project's own continuous-improvement convention
        // (`RUST_LOG=debug`).
        tracing::debug!(
            extra_headers = extra_headers.len(),
            "fetching fresh: {}",
            RedactedUrl::new(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: RedactedUrl::new(url),
            source: e.into(),
        })?;

        if !response.status().is_success() {
            return Err(DepsError::HttpStatus {
                url: RedactedUrl::new(url),
                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)
    }

    /// POSTs `body` as JSON and returns the response body.
    ///
    /// Deliberately does not cache: the OSV batch endpoint is a POST with a
    /// request-body-dependent response and sends no `ETag`/`Last-Modified`
    /// validators, so entry-map caching would be meaningless here — every
    /// call reuses the client, HTTPS enforcement, size cap, and timeout
    /// (via `read_body_capped`) without touching the entry map or
    /// [`Self::total_bytes`].
    ///
    /// # Errors
    ///
    /// Returns `DepsError::HttpStatus` if the server returns a non-2xx
    /// status, `DepsError::RegistryError` if the request fails, or
    /// `DepsError::ResponseTooLarge` if the response body exceeds the
    /// configured size cap.
    #[tracing::instrument(
        skip(self, body),
        fields(url = %RedactedUrl::new(url))
    )]
    pub async fn post_json<T: Serialize + Sync + ?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: RedactedUrl::new(url),
                source: e.into(),
            })?;

        if !response.status().is_success() {
            return Err(DepsError::HttpStatus {
                url: RedactedUrl::new(url),
                status: response.status().as_u16(),
            });
        }

        read_body_capped(url, response, BodyLimit::DEFAULT).await
    }

    /// GETs `url` and returns the response body, bypassing the entry-map
    /// cache entirely — reuses the client, HTTPS enforcement, size cap, and
    /// timeout, exactly like [`Self::post_json`], but for a plain GET.
    ///
    /// For a caller whose own values are already cached elsewhere (e.g.
    /// `OsvClient`'s record cache, validated by a `modified` timestamp
    /// rather than `ETag`/`Last-Modified`): reusing [`Self::get_cached`]
    /// there would double-cache every fetched record in *this* cache's byte
    /// budget too, competing with registry responses for it even though
    /// nothing here ever reads that cached copy back.
    ///
    /// # Errors
    ///
    /// Returns `DepsError::HttpStatus` if the server returns a non-2xx
    /// status, `DepsError::RegistryError` if the request fails, or
    /// `DepsError::ResponseTooLarge` if the response body exceeds the
    /// configured size cap.
    pub async fn get_transport_only(&self, url: &str) -> Result<Bytes> {
        self.get_transport_only_with_headers(url, &[]).await
    }

    /// Same as [`Self::get_transport_only`], but injects extra request headers (e.g. a
    /// content-negotiating `Accept`) — mirrors how [`Self::get_cached_with_headers`] relates
    /// to [`Self::get_cached`].
    ///
    /// # Errors
    ///
    /// Returns `DepsError::HttpStatus` if the server returns a non-2xx
    /// status, `DepsError::RegistryError` if the request fails, or
    /// `DepsError::ResponseTooLarge` if the response body exceeds the
    /// configured size cap.
    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
    }

    /// Same as [`Self::get_transport_only_with_headers`], but takes an explicit
    /// [`BodyLimit`] instead of the [`BodyLimit::DEFAULT`] (`MAX_RESPONSE_BYTES`) cap.
    ///
    /// For a caller whose response is legitimately larger than every other registry
    /// payload — e.g. `deps-pypi`'s full Simple API project index — without weakening
    /// the cap every other caller of this cache relies on. `BodyLimit` clamps at
    /// construction, so this can never be widened past `ABSOLUTE_MAX_RESPONSE_BYTES`
    /// regardless of what the caller passes in.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_transport_only_with_headers`].
    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
    }

    /// Same as [`Self::get_transport_only_with_headers_limited`], but additionally
    /// stops any redirect hop that does not match `trusted_origin` via
    /// [`crate::net_policy::is_trusted_prefix`]
    /// (see [`Self::get_cached_trusted_origin`], which applies the identical policy
    /// to the entry-cached path). For a caller carrying a materially larger
    /// [`BodyLimit`] than [`BodyLimit::DEFAULT`] — the bigger the budget, the more
    /// worth pinning the origin an arbitrary cross-host redirect could point it at.
    ///
    /// # Errors
    ///
    /// Same as [`Self::get_transport_only_with_headers_limited`].
    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
    }

    /// Note: "transport" here means "bypasses the entry-map cache" (see this method's
    /// callers' docs) — a different axis from the [`Transport`] type, which pairs a `Client`
    /// with its [`CacheTier`]. The name predates that type and is kept as-is to avoid
    /// churning every `get_transport_only*` call site for a naming collision that causes no
    /// actual ambiguity at the call sites themselves.
    #[tracing::instrument(
        skip(self, extra_headers, limit, client),
        fields(url = %RedactedUrl::new(url))
    )]
    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: RedactedUrl::new(url),
            source: e.into(),
        })?;

        if !response.status().is_success() {
            return Err(DepsError::HttpStatus {
                url: RedactedUrl::new(url),
                status: response.status().as_u16(),
            });
        }

        read_body_capped(url, response, limit).await
    }

    /// Inserts (or replaces) a cache entry, keeping [`Self::total_bytes`] in sync.
    ///
    /// `DashMap::insert` returns the replaced value, if any, so the byte
    /// delta is computed from a single insert rather than a separate
    /// lookup-then-insert (which would race with concurrent writers).
    ///
    /// A body larger than [`MAX_CACHEABLE_ENTRY_BYTES`] is not inserted at
    /// all (the caller already has it from the network response; only
    /// caching is skipped), and any stale entry previously cached for this
    /// URL is dropped rather than left to serve increasingly outdated data.
    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);
    }

    /// Clears all cached entries.
    ///
    /// This removes all cached responses, forcing the next request for
    /// any URL to fetch fresh data from the network.
    pub fn clear(&self) {
        self.entries.clear();
        self.total_bytes.store(0, Ordering::Relaxed);
    }

    /// Returns the number of cached entries.
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Returns `true` if the cache contains no entries.
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Returns the total bytes retained across all cached response bodies.
    pub fn total_bytes(&self) -> usize {
        self.total_bytes.load(Ordering::Relaxed)
    }

    /// Evicts the oldest cache entries when either capacity limit is reached.
    ///
    /// When the entry count is at or over `MAX_CACHE_ENTRIES`, evicts at
    /// least `CACHE_EVICTION_PERCENTAGE`% of entries (by count). Note this
    /// is a *fix*, not a preserved behavior: the original count-only
    /// eviction built its bounded min-heap with an inverted comparison
    /// (`peek()` returns the oldest entry, but the old code treated it as
    /// the newest-of-the-oldest-so-far and only replaced it when a *newer*
    /// candidate came along that was still older than it — backwards), so
    /// it evicted roughly the first `target_removals` entries in DashMap
    /// hash-iteration order, not the oldest ones. This version evicts
    /// genuinely oldest-first.
    ///
    /// Independently, if the tracked byte total is over [`MAX_CACHE_BYTES`]
    /// — which can happen with far fewer than `MAX_CACHE_ENTRIES` entries if
    /// a few responses are large — eviction keeps removing the next-oldest
    /// entries until the byte budget is satisfied too. A cache that is over
    /// the byte budget but well under the entry-count threshold only evicts
    /// as many entries as the byte budget requires, not a fixed count-based
    /// batch.
    ///
    /// Builds a min-heap over all entry keys by `fetched_at` (O(N)), then
    /// pops the oldest one at a time (O(R log N) for R removals) — unlike a
    /// heap bounded to a fixed top-K, the removal count isn't known upfront
    /// since it depends on the byte budget as well as the count target.
    ///
    /// Every byte-count adjustment here is a relative `fetch_sub` applied to
    /// exactly the entry [`DashMap::remove`] actually returned — never a
    /// snapshot-then-absolute-`store` of a locally computed total. The
    /// latter would silently discard any [`Self::store_entry`] delta that
    /// lands between this method's start and its end (lost-update race), and
    /// under adversarial timing could even underflow `total_bytes` to
    /// `usize::MAX`, permanently wedging every future request into
    /// evicting the entire cache. Reading `total_bytes` fresh on every loop
    /// iteration (rather than maintaining a local mirror) keeps this
    /// correct under concurrent `evict_entries`/`store_entry` calls: two
    /// callers can race to remove the same key — the second `remove` simply
    /// returns `None` and is a no-op, not a double-subtraction.
    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)
        );
    }

    /// Benchmark-only helper: Direct cache lookup without network requests.
    #[cfg(feature = "test-util")]
    #[doc(hidden)]
    pub fn get_for_bench(&self, url: &str) -> Option<Bytes> {
        self.entries.get(url).map(|entry| entry.body.clone())
    }

    /// Benchmark-only helper: Direct cache insertion.
    #[cfg(feature = "test-util")]
    #[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::*;

    use std::assert_matches;

    // Guards the non-loopback path of `ensure_https`: every other test in this
    // module reaches it only through loopback `mockito` URLs, so without this
    // test the "reject any other HTTP host" branch would never run.
    #[test]
    fn test_ensure_https_rejects_non_loopback_http() {
        assert!(ensure_https("http://example.com").is_err());
    }

    /// #767 M2: `ensure_https`'s `CacheError` used to bake the raw rejected URL in
    /// verbatim, the same message-level leak class fixed in `deps-cargo::sparse`'s
    /// fail-closed log on the same `window/showMessage` path.
    #[test]
    fn test_ensure_https_rejection_message_redacts_query_string() {
        let err = ensure_https("http://example.com/pkg?token=super-secret-value").unwrap_err();
        assert!(
            !err.to_string().contains("super-secret-value"),
            "err: {err}"
        );
    }

    // `http://example.com` alone would still pass under a regressed, substring-based
    // `is_loopback_host` (e.g. `url.contains("localhost")`) — these hosts embed a
    // loopback token without actually being loopback, and must still be rejected.
    #[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());
    }

    // The sole exerciser of `is_loopback_host`'s bracketed-IPv6 branch
    // (`strip_prefix('[')`/`split(']')`) — every other test/mockito URL in this
    // module uses `127.0.0.1`.
    #[test]
    fn test_ensure_https_accepts_bracketed_ipv6_loopback() {
        assert!(ensure_https("http://[::1]:1234/x").is_ok());
    }

    // reqwest's `Attempt` has no public constructor, so the redirect closure can't be
    // unit-tested directly; this exercises the pure detection logic it delegates to (mockito is
    // http-only, so an actual https->http redirect isn't testable end-to-end here).
    #[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));
    }

    // The loopback carve-out (identical to `ensure_https`'s) must still exempt
    // loopback hops under `cfg(test)`, or every mockito redirect chain in this
    // module's own tests would start failing.
    #[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));
    }

    // Issue #449: the connect-time resolver guard's pure classification core, unit-tested
    // directly rather than through `tokio::net::lookup_host` — no live DNS/network needed.
    #[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 { .. })
        );
    }

    // FR-003: an attacker's public A record alongside a blocked one must not keep the probe
    // alive — the whole resolution is rejected, not filtered down to the public address.
    #[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
        );
    }

    // NFR-004: fail-closed on an empty resolution rather than silently treating "nothing
    // resolved" as "nothing to block".
    #[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 { .. })
        );
    }

    // Issue #455, test-plan item 1: under `Baseline`, an RFC1918/CGNAT/unique-local address is
    // allowed (today's pre-#455 behavior) — only `never_a_registry` classes are 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"
            );
        }
    }

    // Issue #455, test-plan item 1: under `WorkspaceDeclared(PublicOnly)`, every RFC1918/CGNAT/
    // unique-local address is blocked, while a `Global` address is still allowed.
    #[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-plan item 2: `WorkspaceDeclared(All)` admits a private-range address.
    #[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-plan item 2: `WorkspaceDeclared(Off)` rejects even a `Global` address.
    #[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 { .. })
        );
    }

    // Test-plan item 3: `build_guarded_client_with_lookup` shares `build_client_inner` with the
    // production `build_guarded_client`, so deleting the `.dns_resolver(...)` wiring from that
    // shared function fails this test too, not just the production-path wiring test above. The
    // synthetic lookup returns an RFC1918 address, resolved (not connected — the resolver
    // guard rejects before any TCP attempt) purely through the built `Client`.
    #[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:?}"
        );
    }

    // Test-plan item 3, `Baseline` contrast: the same synthetic private-range lookup is not
    // blocked at the resolver-guard step under `Baseline` — asserted directly against the
    // resolver (not a full `Client`) to avoid depending on any real network behavior of
    // actually connecting to the synthetic address.
    #[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"
        );
    }

    // Direct unit coverage of `BlockedAddrResolver::resolve` on a *name* (not an IP literal —
    // that path never reaches any resolver in production, see the struct's `# Known
    // limitations` doc). `localhost` resolves via the OS's own hosts file, no network needed.
    // This alone does not prove the resolver is wired into `build_guarded_client` — see the
    // sibling test below (critic S1) for that.
    #[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());
    }

    // Issue #449 critic S1: the prior version called `BlockedAddrResolver::resolve` directly and
    // never went through `build_guarded_client`, so deleting `.dns_resolver(...)` from
    // `build_client_inner` left it green. This proves actual wiring: a real mockito listener
    // answers on `server.socket_address()`'s port, reached via the `localhost` *name* so the
    // request actually reaches the configured resolver (unlike an IP literal — see
    // `BlockedAddrResolver`'s `# Known limitations` doc).
    #[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",
        );
        // `Debug` (unlike `Display`) surfaces the boxed source chain, confirming
        // `ResolveGuardError::Blocked` produced the error rather than an unrelated failure.
        let debug = format!("{err:?}");
        assert!(
            debug.contains("Blocked") && debug.contains("Loopback"),
            "expected the failure to originate from ResolveGuardError::Blocked with class \
             Loopback, got: {debug}"
        );
    }

    // S5 (plan-1b §1.1/§4): a 302 to the cloud-metadata IP must be stopped by the
    // *unconditional* redirect policy, not just the trusted-origin one — this is the
    // empirical proof that #443's default unauthenticated client also closes the
    // redirect-hop bypass, not only `get_cached_trusted_origin`.
    #[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");
    }

    // Issue #455, test-plan item 4(a): loopback -> loopback. `Baseline` follows the hop
    // (test-cfg carve-out for `Loopback`); `WorkspaceDeclared(PublicOnly)` stops it since
    // `PublicOnly.allows(Loopback) == false` — the contrast proving the tier split is real.
    #[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(), None)
            .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), None)
            .await;
        assert!(
            matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
            "expected the workspace transport to stop the loopback hop, got {result:?}"
        );
    }

    // Issue #455, test-plan item 4(b): a workspace-blocked literal. The redirect-policy tier
    // term rejects an RFC1918-literal target from its URL string alone (no resolver involved),
    // so the caller sees `HttpStatus{302}` with no `HTTP_TIMEOUT_SECS` stall.
    #[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), None)
            .await;

        assert!(
            matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
            "expected the redirect to 10.0.0.1 to be stopped, got {result:?}"
        );
    }

    // Unlike the https->http downgrade case, cross-origin redirect blocking IS reachable
    // through mockito: two separate `mockito::Server` instances bind to distinct ports,
    // and a distinct port is a distinct origin (scheme+host+port), so a 302 from one to
    // the other is a genuine cross-origin redirect the trusted-origin policy must stop.
    #[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;

        // The stopped redirect surfaces as the 302 response, like any other non-2xx status.
        // `matches!` rather than debug-formatting `result`: the `Ok` arm holds the raw body.
        assert!(
            matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
            "expected HttpStatus(302)"
        );

        // Proves the escape origin was never contacted, not just that the result is a 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");
    }

    /// Issue #795: a raw `str::starts_with` prefix test (the pre-fix behavior) is satisfied
    /// by a subdomain-suffix bypass — `gitlab.mycorp.dev.evil.com` starts with
    /// `https://gitlab.mycorp.dev` as a string, even though its actual host is
    /// `gitlab.mycorp.dev.evil.com`, entirely under attacker control. Parsed-origin equality
    /// must reject it.
    #[test]
    fn test_is_trusted_origin_rejects_subdomain_suffix_bypass() {
        let trusted = Url::parse("https://gitlab.mycorp.dev").unwrap();
        let hop = Url::parse("https://gitlab.mycorp.dev.evil.com/steal").unwrap();
        assert!(!is_trusted_origin(&hop, Some(&trusted)));
    }

    /// Issue #795: a userinfo bypass — `https://gitlab.mycorp.dev@evil.com/...` also starts
    /// with the trusted origin as a string, but its host is `evil.com`; `gitlab.mycorp.dev`
    /// is merely a (discarded) username. `Url::origin()` ignores userinfo entirely, so this
    /// must be rejected.
    #[test]
    fn test_is_trusted_origin_rejects_userinfo_bypass() {
        let trusted = Url::parse("https://gitlab.mycorp.dev").unwrap();
        let hop = Url::parse("https://gitlab.mycorp.dev@evil.com/steal").unwrap();
        assert!(!is_trusted_origin(&hop, Some(&trusted)));
    }

    /// Issue #795: a hyphen-suffix bypass — `gitlab.mycorp.dev-evil.com` again starts with
    /// the trusted origin as a string while being an entirely distinct, attacker-controlled
    /// host.
    #[test]
    fn test_is_trusted_origin_rejects_hyphen_suffix_bypass() {
        let trusted = Url::parse("https://gitlab.mycorp.dev").unwrap();
        let hop = Url::parse("https://gitlab.mycorp.dev-evil.com/steal").unwrap();
        assert!(!is_trusted_origin(&hop, Some(&trusted)));
    }

    /// Companion to the three bypass-rejection tests above: the legitimate same-origin case
    /// (a different path, same scheme/host/port) must still be accepted.
    #[test]
    fn test_is_trusted_origin_accepts_exact_origin_match() {
        let trusted = Url::parse("https://gitlab.mycorp.dev").unwrap();
        let hop = Url::parse("https://gitlab.mycorp.dev/api/v4/x").unwrap();
        assert!(is_trusted_origin(&hop, Some(&trusted)));
    }

    /// A `trusted_origin` that fails to parse must fail closed — every hop is rejected,
    /// never treated as "no restriction".
    #[test]
    fn test_is_trusted_origin_rejects_when_trusted_origin_unparseable() {
        let hop = Url::parse("https://gitlab.mycorp.dev/api/v4/x").unwrap();
        assert!(!is_trusted_origin(&hop, None));
    }

    /// Path-prefix scoping (NuGet's registration-hive/flat-container pinning) must survive
    /// the #795 origin-equality fix: same origin, but a hop outside the trusted path, is
    /// still rejected — this is what `test_get_cached_trusted_origin_rejects_sibling_path_prefix`
    /// exercises end-to-end; this is the same property pinned at the unit level.
    #[test]
    fn test_is_trusted_origin_rejects_same_origin_sibling_path() {
        let trusted = Url::parse("https://registry.example/v3/registration5-gz/").unwrap();
        let hop = Url::parse("https://registry.example/v3/registration5-gzX/evil").unwrap();
        assert!(!is_trusted_origin(&hop, Some(&trusted)));
    }

    /// Companion: same origin, hop path under the trusted path prefix, is still accepted.
    #[test]
    fn test_is_trusted_origin_accepts_same_origin_nested_path() {
        let trusted = Url::parse("https://registry.example/v3/registration5-gz/").unwrap();
        let hop =
            Url::parse("https://registry.example/v3/registration5-gz/serde/page1.json").unwrap();
        assert!(is_trusted_origin(&hop, Some(&trusted)));
    }

    /// Issue #795 S1: unlike the two trailing-slash tests above (which, with a trailing `/`
    /// already present in the trusted path, would have passed even under the pre-S1-fix raw
    /// `str::starts_with` check — they do not actually exercise the segment-boundary fix),
    /// `RegistryIndex::as_str()` (`deps-cargo`'s sparse-index trusted origin) carries **no**
    /// trailing-slash guarantee. This reproduces that exact shape and the critic's repro: a
    /// same-origin sibling whose path merely shares a textual prefix must still be rejected.
    #[test]
    fn test_is_trusted_origin_rejects_same_origin_sibling_path_no_trailing_slash() {
        let trusted = Url::parse("https://artifacts.corp/cargo/index").unwrap();
        for sibling in [
            "https://artifacts.corp/cargo/index-public/steal",
            "https://artifacts.corp/cargo/indexEVIL",
            "https://artifacts.corp/cargo/index.evil/x",
        ] {
            let hop = Url::parse(sibling).unwrap();
            assert!(
                !is_trusted_origin(&hop, Some(&trusted)),
                "expected {sibling} to be rejected"
            );
        }
    }

    /// Companion: the trusted path itself, and a proper child path, are still accepted when
    /// the trusted path carries no trailing slash — the real `deps-cargo` request shape
    /// (`sparse_index_url` appends `/{crate_path}` to the trimmed base).
    #[test]
    fn test_is_trusted_origin_accepts_self_and_child_no_trailing_slash() {
        let trusted = Url::parse("https://artifacts.corp/cargo/index").unwrap();
        let itself = Url::parse("https://artifacts.corp/cargo/index").unwrap();
        let child = Url::parse("https://artifacts.corp/cargo/index/se/rd/serde").unwrap();
        assert!(is_trusted_origin(&itself, Some(&trusted)));
        assert!(is_trusted_origin(&child, Some(&trusted)));
    }

    // Proves every hop is re-checked, not just the first: a same-origin hop is followed,
    // then a second, cross-origin hop from that (already-followed) intermediate is stopped.
    #[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;

        // matches! rather than debug-formatting result: the Ok arm holds the raw body.
        assert!(
            matches!(result, Err(DepsError::HttpStatus { status: 302, .. })),
            "expected HttpStatus(302)"
        );
        escape.assert_async().await;
    }

    // Sibling path-prefix rejection: `.../api/` must not accept `.../apiX/...`. The other
    // trusted-origin tests use a bare-host prefix, which never exercises this boundary.
    #[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;

        // matches! rather than debug-formatting result: the Ok arm holds the raw body.
        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");
    }

    // A credential header must never survive a cross-origin redirect hop — proven by the
    // escape origin never being contacted, not just the header being absent on a landed request.
    #[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;
    }

    // Proves `redirect_policy`'s delegation is actually wired in and live: without it, this
    // chain would keep following past 10 hops instead of erroring — a single-hop test can't
    // distinguish "delegation is live" from "no policy at all".
    #[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();

        // reqwest errors once `previous.len() > 10` (the 11th hop), so 11 redirecting steps
        // (step/0..step/10) are needed to trigger it; step/11 must never be requested.
        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;

        // Kept alive until here: each `Mock` deregisters on drop, turning hops 404 otherwise.
        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();
        // Bytes clone is cheap (reference counting)
        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();
        // https:// (not http://) so this exercises DNS-resolution failure, not the
        // HTTPS-only policy enforced by `ensure_https`.
        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())
        );
    }

    /// #756 security follow-up S-A: the "fetching fresh: {url}" debug log in
    /// `fetch_and_store_with_headers` — the direct callee `get_cached_with_headers_via`'s
    /// `miss` branch delegates to — must never carry a token embedded in the URL's query
    /// string. This is `debug`-level, the level this project's own continuous-improvement
    /// convention runs at (`RUST_LOG=debug`), so it is not merely a theoretical exposure.
    #[cfg(feature = "test-util")]
    #[tokio::test]
    async fn test_fetch_and_store_fetching_fresh_log_redacts_query_string_token() {
        let mut server = mockito::Server::new_async().await;
        let url = format!("{}/pkg?token=super-secret-value", server.url());

        let _m = server
            .mock("GET", "/pkg")
            .match_query(mockito::Matcher::UrlEncoded(
                "token".into(),
                "super-secret-value".into(),
            ))
            .with_status(200)
            .with_body("ok")
            .create_async()
            .await;

        let cache = HttpCache::new();
        let output =
            crate::test_util::capture_tracing_output_async_at(tracing::Level::DEBUG, async {
                let result: Bytes = cache
                    .fetch_and_store_with_headers(&url, &[], &cache.baseline.client, &url)
                    .await
                    .unwrap();
                assert_eq!(result.as_ref(), b"ok");
            })
            .await;

        assert!(
            !output.contains("super-secret-value"),
            "leaked token via 'fetching fresh' debug log: {output:?}"
        );
    }

    /// #767 S3 / #789: proves `SanitizedRegistryError`'s `From<reqwest::Error>` actually
    /// strips the URL from the wrapped `reqwest::Error`'s own `Display`, not just that
    /// `RegistryError::package` is redacted — a genuine transport-level error (connection
    /// refused on a closed loopback port, so `.url()` is populated the way a builder-only
    /// error like `Client::get("not a url").build().unwrap_err()` never is) is required to
    /// exercise this: reverting any of the 5 `source: e.into()` call sites in this file back
    /// to a bare `reqwest::Error` must fail to compile, and reverting
    /// `SanitizedRegistryError::from`'s `.without_url()` call must fail this test.
    #[tokio::test]
    async fn test_registry_error_source_redacts_url_on_real_transport_error() {
        // Bind then immediately drop a loopback listener: nothing accepts connections on
        // this port afterward, so a request to it fails fast with connection-refused
        // instead of hanging or needing a real unreachable host.
        let port = {
            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            listener.local_addr().unwrap().port()
        };
        let url = format!("http://127.0.0.1:{port}/pkg?token=super-secret-value");

        let cache = HttpCache::new();
        let err = cache.get_cached(&url).await.unwrap_err();

        assert_matches!(err, DepsError::RegistryError { .. });
        assert!(
            !err.to_string().contains("super-secret-value"),
            "err: {err}"
        );
    }

    #[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");
    }

    /// The headered form of `get_cached_workspace` used by `deps-npm`'s alternate-registry
    /// client (A1): forwards `extra_headers` while still going through the workspace-tier
    /// transport (mirrors `test_get_cached_and_get_cached_workspace_do_not_share_an_entry`'s
    /// use of the unheadered `get_cached_workspace` against a loopback mockito server under
    /// the default policy — an IP-literal host like mockito's has no DNS resolution step for
    /// the connect-time `AddrGuard` to intercept, so no policy elevation is needed here
    /// either; that guard's actual job is catching a *hostname* that resolves differently at
    /// connect time than its parse-time classification, see `validate_resolved_addrs`).
    #[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() {
        // MAX_RESPONSE_BYTES is well over MAX_CACHEABLE_ENTRY_BYTES, so the network-layer cap
        // and the cache admission cap are independent (see test_store_entry_skips_caching_oversized_entry).
        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(),
            },
        );

        // Registry down for maintenance: a non-2xx, non-304 response instead of "unchanged"
        // or "here's the new body".
        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();

        // Stale-while-revalidate: last-known-good body returned, entry untouched.
        assert_eq!(result.as_ref(), b"stale but good");
        let cached = cache.entries.get(&url).unwrap();
        assert_eq!(cached.etag, Some("\"stale-etag\"".into()));
    }

    /// #756 round 2 S1 regression: the "conditional request failed, using cache" warn (fired
    /// on exactly this stale-while-revalidate path) must never interpolate the `DepsError`
    /// itself — `DepsError::HttpStatus`'s `Display` embeds the full, unredacted URL (including
    /// the query string), which would defeat `RedactedUrl`'s redaction on this same span's
    /// `url` field two lines above it. Reuses the mock/seeding shape of
    /// `test_get_cached_non_2xx_on_refresh_preserves_stale_cache` with a token-bearing query
    /// string, wrapped in a real tracing capture (an actual `warn!` event fires here, unlike
    /// the offline-hit case covered by `test_get_cached_span_url_field_redacts_query_string_token`).
    #[cfg(feature = "test-util")]
    #[tokio::test]
    async fn test_get_cached_conditional_request_failure_does_not_leak_token_via_warn() {
        let mut server = mockito::Server::new_async().await;
        let url = format!("{}/pkg?token=super-secret-value", 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", "/pkg")
            .match_query(mockito::Matcher::UrlEncoded(
                "token".into(),
                "super-secret-value".into(),
            ))
            .match_header("if-none-match", "\"stale-etag\"")
            .with_status(503)
            .with_body("<html>maintenance</html>")
            .create_async()
            .await;

        let output = crate::test_util::capture_tracing_output_async(async {
            let result: Bytes = cache.get_cached(&url).await.unwrap();
            assert_eq!(result.as_ref(), b"stale but good");
        })
        .await;

        assert!(
            !output.contains("super-secret-value"),
            "leaked token via warn! output: {output:?}"
        );
    }

    /// #756 C1 regression: `get_cached_with_headers_via`'s `url` span field must never carry
    /// a token embedded in the URL's query string — the same shape as an `.npmrc`-style
    /// `${VAR}`-expanded `registry=` URL (see `deps-npm`'s `NpmRegistryIndex` security model).
    /// The critic's exact repro was the span's own `url={...}` prefix on a captured log line,
    /// so this enables `FmtSpan::NEW` to capture that prefix directly, at span-creation time
    /// (before the function body runs), rather than relying on some other event firing inside
    /// the span. Deliberately exercises the offline-hit branch — no `DepsError` is ever
    /// constructed on that path — so this is isolated from that type's own (separate,
    /// pre-existing) URL-embedding `Display` impl and tests the span field in isolation.
    #[cfg(feature = "test-util")]
    #[tokio::test]
    async fn test_get_cached_span_url_field_redacts_query_string_token() {
        use tracing_subscriber::fmt::format::FmtSpan;

        #[derive(Clone)]
        struct TestWriter(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
        impl std::io::Write for TestWriter {
            fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
                self.0.lock().unwrap().extend_from_slice(data);
                Ok(data.len())
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }
        impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestWriter {
            type Writer = Self;
            fn make_writer(&'a self) -> Self::Writer {
                self.clone()
            }
        }

        let url = "https://npm.internal/pkg?token=super-secret-value".to_string();
        let cache = HttpCache::new();
        cache.set_offline(true);
        cache.entries.insert(
            url.clone(),
            CachedResponse {
                body: Bytes::from_static(b"cached"),
                etag: None,
                last_modified: None,
                fetched_at: Instant::now(),
            },
        );

        // TODO(critic): migrate onto deps_core::test_util capture helpers once they expose
        // span events (#1006).
        let buf = std::sync::Arc::new(std::sync::Mutex::new(Vec::<u8>::new()));
        let subscriber = tracing_subscriber::fmt()
            .with_writer(TestWriter(std::sync::Arc::clone(&buf)))
            .with_span_events(FmtSpan::NEW)
            .without_time()
            .with_target(false)
            .finish();

        let guard = tracing::subscriber::set_default(subscriber);
        let result: Bytes = cache.get_cached(&url).await.unwrap();
        drop(guard);

        assert_eq!(result.as_ref(), b"cached");

        let output = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        assert!(
            !output.contains("super-secret-value"),
            "leaked token into span output: {output:?}"
        );
        assert!(
            output.contains("https://npm.internal/pkg"),
            "expected the redacted host+path in span output: {output:?}"
        );
    }

    #[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);

        // Replacing the same key must account for the delta, not just add.
        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();

        // 9 entries at the per-entry admission cap: far below MAX_CACHE_ENTRIES by count, but
        // their combined size (72 MiB) overshoots MAX_CACHE_BYTES (64 MiB).
        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();

        // Only as many oldest entries as needed to clear the byte budget are removed, not a
        // fixed count-based batch.
        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();

        // Evicting just the single oldest entry should restore the cache to within budget,
        // proving eviction picks the genuinely oldest entry, not hash-iteration order.
        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();

        // Pre-fill past the byte budget, staying under MAX_CACHE_ENTRIES by count.
        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");

        // The pre-request byte-budget check evicted stale entries before fetching.
        assert!(cache.total_bytes() <= MAX_CACHE_BYTES + result.len());
    }

    #[test]
    fn test_store_entry_skips_caching_oversized_entry() {
        let cache = HttpCache::new();

        // A body over the per-entry admission cap is not retained, even though the caller
        // still gets it back (store_entry's caller already holds `body` independently).
        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);

        // Replacing an existing small entry with an oversized one drops the stale entry too,
        // rather than leaving it to keep serving increasingly outdated data.
        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();

        // Regression guard: evict_entries used to snapshot total_bytes once and overwrite it
        // with an absolute store, silently discarding any concurrent store_entry delta.
        let actual: usize = cache
            .entries
            .iter()
            .map(|entry| entry.value().body.len())
            .sum();
        assert_eq!(cache.total_bytes(), actual);
    }

    // Issue #455, test-plan item 5: `get_cached(url)` then `get_cached_workspace(url)` do not
    // share an entry — each is keyed under a distinct namespace (see `HttpCache::cache_key`),
    // so the mockito mock is hit twice and the cache ends up with two entries for one URL.
    #[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;
    }

    // Issue #455, test-plan item 6 (C5): fetch under `All`, tighten to `PublicOnly`, re-fetch
    // the same URL — the `All`-era body must not be served, since the policy-scoped key
    // namespace changes with the policy.
    #[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"
        );
        // mockito's second mock answers regardless of which entry was hit, so the body
        // assertion alone wouldn't prove a policy-blind key — this length check does.
        assert_eq!(cache.len(), 2);
    }

    // Issue #455, test-plan item 7 (C4): `set_registry_policy` rebuilds the workspace transport
    // only on an actual change, not on a no-op re-application of the same value.
    #[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);
    }

    // Issue #483: `.expect(0)` proves nothing reached *this mock*, not that zero sockets
    // ever opened — adequate only because the loopback carve-out lets mockito stand in here.

    #[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, RedactedUrl::new(&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;
    }

    // S1 fix (critic-corrected): the naive design's `!cache_enabled` bypass ran *before*
    // any offline check, so `cache.enabled: false` + `network.offline: true` cold always
    // took the network-only bypass path — which `ensure_online` then blocked — even though
    // this combination is meant to still surface a clean, immediate signal rather than
    // hang or silently return empty data forever.
    #[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;
    }

    // The scenario S1 actually exists to fix: an entry stored while caching was enabled
    // must still be servable once `cache.enabled` is later turned off *and* the cache goes
    // offline in the same breath — proving `offline` truly overrides `cache_enabled` on the
    // read path, not just when the two flags never change together.
    #[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;
    }

    // The primary UX case (critic M6a): a full online -> offline transition on an
    // otherwise-default cache (cache.enabled stays true throughout) must keep serving what
    // was already fetched.
    #[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;
    }

    // Offline -> online restores live fetches (the flag's other half of critic M6a),
    // exercised here through `get_cached`'s conditional-revalidation path directly (the
    // live `did_change_configuration` toggle is covered by `deps-lsp`'s own test).
    #[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;
    }

    // Critic M6b: `get_cached_workspace` and `get_cached_trusted_origin` key entries under
    // distinct namespaces from `get_cached`'s baseline tier — the offline warm-cache path
    // needs its own proof it holds for each.
    #[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;
    }

    // --- issue #561/#562: CacheTier::Pinned, get_cached_pinned{,_with_headers} ---

    #[tokio::test]
    async fn test_get_cached_pinned_attaches_auth_header() {
        let mut server = mockito::Server::new_async().await;
        let trusted_origin = format!("{}/", server.url());
        let url = format!("{}/api/data", server.url());

        let _m = server
            .mock("GET", "/api/data")
            .match_header("authorization", "Basic dXNlcjpwYXQ=")
            .with_status(200)
            .with_body("authenticated data")
            .create_async()
            .await;

        let cache = HttpCache::new();
        let result = cache
            .get_cached_pinned_with_headers(
                &url,
                &trusted_origin,
                true,
                Some(42),
                &[(header::AUTHORIZATION, "Basic dXNlcjpwYXQ=")],
            )
            .await
            .unwrap();

        assert_eq!(result.as_ref(), b"authenticated data");
    }

    /// FR-014: distinct `auth_id` values against the same `(url, trusted_origin)` never share
    /// a cache entry — a rotated or distinct credential never reads back a body fetched under a
    /// different one.
    #[tokio::test]
    async fn test_get_cached_pinned_distinct_auth_id_never_shares_cache_entry() {
        let mut server = mockito::Server::new_async().await;
        let trusted_origin = format!("{}/", server.url());
        let url = format!("{}/api/data", server.url());

        let _m1 = server
            .mock("GET", "/api/data")
            .with_status(200)
            .with_body("body-for-credential-a")
            .create_async()
            .await;

        let cache = HttpCache::new();
        let a = cache
            .get_cached_pinned(&url, &trusted_origin, true, Some(1))
            .await
            .unwrap();
        assert_eq!(a.as_ref(), b"body-for-credential-a");
        drop(_m1);

        let _m2 = server
            .mock("GET", "/api/data")
            .with_status(200)
            .with_body("body-for-credential-b")
            .create_async()
            .await;

        let b = cache
            .get_cached_pinned(&url, &trusted_origin, true, Some(2))
            .await
            .unwrap();
        assert_eq!(
            b.as_ref(),
            b"body-for-credential-b",
            "a distinct auth_id must not read back credential A's cached body"
        );
    }

    /// #1025: `auth_id: None` (unauthenticated) and `auth_id: Some(0)` (a credential whose
    /// salted digest happens to hash to exactly `0`) must produce different `Pinned`-tier
    /// cache keys — collapsing `None` to the same `0` sentinel `Some(0)` digests to would let
    /// a cached authenticated body be served to a subsequent unauthenticated request, or vice
    /// versa.
    #[test]
    fn test_cache_key_pinned_none_and_some_zero_auth_id_never_collide() {
        let cache = HttpCache::new();
        let tier = CacheTier::Pinned {
            digest: 42,
            authenticated: true,
        };

        let unauthenticated = cache.cache_key("https://example.com/pkg", tier, None);
        let zero_digest_credential = cache.cache_key("https://example.com/pkg", tier, Some(0));

        assert_ne!(unauthenticated, zero_digest_credential);
    }

    /// #1025 regression guard: non-zero `auth_id` values keep producing the pre-existing,
    /// distinct-per-id cache key behavior (only the `None` vs `Some(0)` collision was fixed).
    #[test]
    fn test_cache_key_pinned_nonzero_auth_id_still_distinct() {
        let cache = HttpCache::new();
        let tier = CacheTier::Pinned {
            digest: 42,
            authenticated: true,
        };

        let a = cache.cache_key("https://example.com/pkg", tier, Some(1));
        let b = cache.cache_key("https://example.com/pkg", tier, Some(2));
        let unauthenticated = cache.cache_key("https://example.com/pkg", tier, None);

        assert_ne!(a, b);
        assert_ne!(a, unauthenticated);
        assert_ne!(b, unauthenticated);
    }

    /// #1025 M2: proves the fixed-width invariant itself, not just a couple of hand-picked
    /// values — a regression that dropped zero-padding (e.g. `{digest:x}` instead of
    /// `{digest:016x}`) would still pass the two tests above (`None != Some(0)`,
    /// `Some(1) != Some(2)`) but would let differently-shaped `(digest, auth_id)` pairs collide,
    /// e.g. `digest=0x1, auth_id=Some(0x11)` vs. `digest=0x11, auth_id=Some(0x1)` concatenating
    /// to the same string once padding is gone. Every `(digest, auth_id)` combination drawn
    /// from these boundary-value sets must produce a pairwise-distinct key.
    #[test]
    fn test_cache_key_pinned_digest_and_auth_id_matrix_never_collide() {
        let cache = HttpCache::new();
        let digests = [0u64, 1, 0x11, u64::MAX];
        let auth_ids = [None, Some(0u64), Some(1), Some(0x11), Some(u64::MAX)];

        let mut keys = Vec::new();
        for &digest in &digests {
            for &auth_id in &auth_ids {
                let tier = CacheTier::Pinned {
                    digest,
                    authenticated: true,
                };
                keys.push((
                    (digest, auth_id),
                    cache
                        .cache_key("https://example.com/pkg", tier, auth_id)
                        .into_owned(),
                ));
            }
        }

        for i in 0..keys.len() {
            for j in (i + 1)..keys.len() {
                assert_ne!(
                    keys[i].1, keys[j].1,
                    "inputs {:?} and {:?} produced the same cache key",
                    keys[i].0, keys[j].0
                );
            }
        }
    }

    /// #1025 M1: `authenticated` is folded into the `Pinned`-tier cache key too, not just
    /// `auth_id` — otherwise `(authenticated: true, auth_id: None)` and
    /// `(authenticated: false, auth_id: None)` against the same origin would collide. Not
    /// reachable through any current caller (every ecosystem correlates the two), but
    /// `get_cached_pinned_with_headers` is public API with no enforced invariant tying them
    /// together, so the key format must not assume one.
    #[test]
    fn test_cache_key_pinned_authenticated_flag_never_collides_with_unauthenticated() {
        let cache = HttpCache::new();
        let authenticated_tier = CacheTier::Pinned {
            digest: 42,
            authenticated: true,
        };
        let unauthenticated_tier = CacheTier::Pinned {
            digest: 42,
            authenticated: false,
        };

        let a = cache.cache_key("https://example.com/pkg", authenticated_tier, None);
        let b = cache.cache_key("https://example.com/pkg", unauthenticated_tier, None);

        assert_ne!(a, b);
    }

    /// FR-015/NFR-004: a 401 revalidation response against an authenticated `Pinned`-tier
    /// entry evicts the entry and returns the error — never the default
    /// stale-while-revalidate fallback that would serve the possibly-revoked credential's
    /// last-known-good body.
    #[tokio::test]
    async fn test_pinned_authenticated_401_revalidation_evicts_instead_of_stale_serve() {
        let mut server = mockito::Server::new_async().await;
        let trusted_origin = format!("{}/", server.url());
        let url = format!("{}/api/data", server.url());

        let _m1 = server
            .mock("GET", "/api/data")
            .with_status(200)
            .with_header("etag", "\"abc123\"")
            .with_body("private data")
            .create_async()
            .await;

        let cache = HttpCache::new();
        let first = cache
            .get_cached_pinned(&url, &trusted_origin, true, Some(7))
            .await
            .unwrap();
        assert_eq!(first.as_ref(), b"private data");
        assert_eq!(cache.len(), 1);
        drop(_m1);

        let _m2 = server
            .mock("GET", "/api/data")
            .match_header("if-none-match", "\"abc123\"")
            .with_status(401)
            .create_async()
            .await;

        let result = cache
            .get_cached_pinned(&url, &trusted_origin, true, Some(7))
            .await;

        assert!(
            matches!(result, Err(DepsError::HttpStatus { status: 401, .. })),
            "expected the 401 to surface as an error, not a stale-served body: {result:?}"
        );
        assert_eq!(
            cache.len(),
            0,
            "the revoked-credential entry must be evicted, not left cached"
        );
    }

    /// Every other tier keeps today's stale-while-revalidate fallback unchanged — only an
    /// *authenticated* `Pinned` entry evicts on 401/403 (FR-015's scope is deliberately
    /// narrow).
    #[tokio::test]
    async fn test_unauthenticated_pinned_401_revalidation_still_serves_stale() {
        let mut server = mockito::Server::new_async().await;
        let trusted_origin = format!("{}/", server.url());
        let url = format!("{}/api/data", server.url());

        let _m1 = server
            .mock("GET", "/api/data")
            .with_status(200)
            .with_header("etag", "\"abc123\"")
            .with_body("workspace data")
            .create_async()
            .await;

        let cache = HttpCache::new();
        let first = cache
            .get_cached_pinned(&url, &trusted_origin, false, None)
            .await
            .unwrap();
        assert_eq!(first.as_ref(), b"workspace data");
        drop(_m1);

        let _m2 = server
            .mock("GET", "/api/data")
            .match_header("if-none-match", "\"abc123\"")
            .with_status(401)
            .create_async()
            .await;

        let second = cache
            .get_cached_pinned(&url, &trusted_origin, false, None)
            .await
            .unwrap();
        assert_eq!(
            second.as_ref(),
            b"workspace data",
            "an unauthenticated Pinned entry must keep the default stale-while-revalidate fallback"
        );
        assert_eq!(cache.len(), 1);
    }

    /// `set_registry_policy` purges every `Pinned`-tier cache entry (and pooled transport) on
    /// an actual policy transition — closing the round-trip hole for credential-carrying
    /// entries (NFR-004), unlike the pre-existing `WorkspaceDeclared` non-purge behavior.
    #[tokio::test]
    async fn test_set_registry_policy_purges_pinned_tier_entries() {
        let mut server = mockito::Server::new_async().await;
        let trusted_origin = format!("{}/", server.url());
        let url = format!("{}/api/data", server.url());

        let _m = server
            .mock("GET", "/api/data")
            .with_status(200)
            .with_body("private data")
            .create_async()
            .await;

        let policy = Arc::new(RegistryAccessPolicy::new(WorkspaceRegistryAccess::All));
        let cache = HttpCache::with_policy(Arc::clone(&policy));
        cache
            .get_cached_pinned(&url, &trusted_origin, true, Some(1))
            .await
            .unwrap();
        assert_eq!(cache.len(), 1);

        cache.set_registry_policy(WorkspaceRegistryAccess::PublicOnly);

        assert_eq!(
            cache.len(),
            0,
            "a Pinned-tier entry must be purged on any actual policy transition"
        );
    }
}