holger-server-lib 0.6.9

//! Read-only **remote Cargo registry** upstream (default: crates.io).
//!
//! Used as a [`ProxyBackend`](crate::proxy::ProxyBackend) upstream so a writable
//! `/cache` primary pulls + caches crates on demand (a transparent pull-through
//! cache). HTTP is SYNC via `ureq` — runtime-agnostic, so it is safe both inside
//! holger's async hyper handler (where blocking on `reqwest` would panic) and in
//! plain unit tests.

use std::io::Read;
use std::sync::OnceLock;
use std::time::Duration;

use traits::{ArtifactFormat, ArtifactId, RepositoryBackendTrait};

/// Hard cap on a single crates.io-mirror download (bytes). A malicious/MITM
/// upstream can otherwise omit/lie about Content-Length and stream an unbounded
/// body into memory, OOMing the server (each fetch runs on a blocking thread).
/// The crates.io mirror is part of holger's core (always compiled), independent
/// of the `federation` tier — so the cap lives here, not in `crate::upstream`.
pub(crate) const MAX_DOWNLOAD_BYTES: u64 = 8 * 1024 * 1024 * 1024;

/// Shared `ureq` agent with bounded timeouts. A naked `ureq::get(url).call()`
/// has **no** connect/read timeout, so a hung or slow upstream (a stalled mirror,
/// a black-holed connection) blocks the calling thread forever. Each upstream
/// fetch runs in `spawn_blocking`, so a handful of stuck calls would pin the
/// blocking pool and starve unrelated repos. Bound connect + read + write so a
/// dead upstream fails fast instead. One agent (connection pool) shared process-
/// wide; built lazily.
pub(crate) fn agent() -> &'static ureq::Agent {
    static AGENT: OnceLock<ureq::Agent> = OnceLock::new();
    AGENT.get_or_init(|| {
        ureq::AgentBuilder::new()
            .timeout_connect(Duration::from_secs(10))
            .timeout_read(Duration::from_secs(30))
            .timeout_write(Duration::from_secs(30))
            .build()
    })
}

/// A read-only proxy to a remote sparse Cargo registry.
pub struct RustRegistryUpstream {
    name: String,
    /// Sparse index base, e.g. `https://index.crates.io`.
    index_base: String,
    /// Crate download base, e.g. `https://static.crates.io/crates`.
    dl_base: String,
}

impl RustRegistryUpstream {
    /// crates.io defaults: the sparse index + the static CDN.
    pub fn crates_io(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            index_base: "https://index.crates.io".to_string(),
            dl_base: "https://static.crates.io/crates".to_string(),
        }
    }

    /// Point at an arbitrary sparse registry (for tests / private mirrors).
    pub fn new(
        name: impl Into<String>,
        index_base: impl Into<String>,
        dl_base: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            index_base: index_base.into(),
            dl_base: dl_base.into(),
        }
    }

    /// Download URL for one `.crate` (the cargo `dl` convention).
    fn crate_url(&self, name: &str, version: &str) -> String {
        format!("{}/{name}/{name}-{version}.crate", self.dl_base)
    }

    /// Blocking GET. `Ok(None)` on 404, `Ok(Some(bytes))` on 200, `Err` otherwise.
    fn http_get(url: &str) -> anyhow::Result<Option<Vec<u8>>> {
        Self::http_get_with(agent(), url)
    }

    /// As [`http_get`](Self::http_get) but against a caller-supplied agent — lets
    /// tests drive a short-timeout agent at a hanging server.
    fn http_get_with(agent: &ureq::Agent, url: &str) -> anyhow::Result<Option<Vec<u8>>> {
        match agent.get(url).call() {
            Ok(resp) => {
                let mut buf = Vec::new();
                // Cap the read: a malicious/MITM upstream can otherwise omit/lie about
                // Content-Length and stream an unbounded body into memory, OOMing the
                // server (each fetch runs on a blocking thread). Mirrors upstream::get_bytes.
                resp.into_reader()
                    .take(MAX_DOWNLOAD_BYTES)
                    .read_to_end(&mut buf)
                    .map_err(|e| anyhow::anyhow!("read upstream body {url}: {e}"))?;
                Ok(Some(buf))
            }
            Err(ureq::Error::Status(404, _)) => Ok(None),
            Err(ureq::Error::Status(code, _)) => anyhow::bail!("upstream {url}: HTTP {code}"),
            Err(e) => Err(anyhow::anyhow!("upstream {url}: {e}")),
        }
    }
}

impl RepositoryBackendTrait for RustRegistryUpstream {
    fn name(&self) -> &str {
        &self.name
    }

    fn format(&self) -> ArtifactFormat {
        ArtifactFormat::Rust
    }

    fn is_writable(&self) -> bool {
        false
    }

    fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<Vec<u8>>> {
        Self::http_get(&self.crate_url(&id.name, &id.version))
    }

    fn put(&self, _id: &ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
        anyhow::bail!("remote registry upstream is read-only")
    }

    fn handle_http2_request(
        &self,
        _method: &str,
        suburl: &str,
        _body: &[u8],
    ) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
        let parts: Vec<&str> = suburl.trim_start_matches('/').split('/').collect();
        match parts.as_slice() {
            // Sparse index metadata: /{repo}/index/{path...}. cargo has already
            // encoded the correct prefix (1/2/3-char rules), so forward verbatim.
            [_repo, "index", rest @ ..] if !rest.is_empty() && rest != ["config.json"] => {
                let url = format!("{}/{}", self.index_base, rest.join("/"));
                match Self::http_get(&url)? {
                    Some(body) => Ok((
                        200,
                        vec![("Content-Type".into(), "text/plain".into())],
                        body,
                    )),
                    None => Ok((404, Vec::new(), b"Not found upstream".to_vec())),
                }
            }
            // Crate download: /{repo}/crates/{name}/{version}/download.
            [_repo, "crates", name, version, "download"] => {
                match Self::http_get(&self.crate_url(name, version))? {
                    Some(body) => Ok((
                        200,
                        vec![("Content-Type".into(), "application/octet-stream".into())],
                        body,
                    )),
                    None => Ok((404, Vec::new(), b"Crate not found upstream".to_vec())),
                }
            }
            _ => Ok((404, Vec::new(), b"Not found".to_vec())),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn crates_io_defaults_build_expected_urls() {
        let up = RustRegistryUpstream::crates_io("crates-io");
        assert_eq!(up.name(), "crates-io");
        assert!(!up.is_writable());
        assert_eq!(
            up.crate_url("serde", "1.0.0"),
            "https://static.crates.io/crates/serde/serde-1.0.0.crate"
        );
    }

    #[test]
    fn custom_base_urls_are_honored() {
        let up = RustRegistryUpstream::new("mirror", "https://idx.example", "https://dl.example/c");
        assert_eq!(
            up.crate_url("tokio", "1.40.0"),
            "https://dl.example/c/tokio/tokio-1.40.0.crate"
        );
    }

    #[test]
    fn put_is_rejected_read_only() {
        let up = RustRegistryUpstream::crates_io("crates-io");
        let id = ArtifactId { namespace: None, name: "x".into(), version: "1".into() };
        assert!(up.put(&id, b"data").is_err(), "upstream must be read-only");
    }

    /// A hung upstream (accepts the TCP connection, then never sends a response)
    /// must make the GET fail on the read timeout instead of blocking forever.
    /// Drives a short-timeout agent at a black-hole listener and asserts the call
    /// returns an `Err` well within a bound — the regression guard for the
    /// previously-unbounded `ureq::get(url).call()`.
    #[test]
    fn hung_upstream_times_out_instead_of_hanging() {
        use std::io::Read as _;
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        // Accept one connection and sit on it (read but never reply) so the
        // client blocks in the read phase until its timeout fires.
        let handle = std::thread::spawn(move || {
            if let Ok((mut sock, _)) = listener.accept() {
                let mut scratch = [0u8; 64];
                let _ = sock.read(&mut scratch);
                std::thread::sleep(Duration::from_secs(2));
            }
        });

        let agent = ureq::AgentBuilder::new()
            .timeout_connect(Duration::from_millis(500))
            .timeout_read(Duration::from_millis(500))
            .build();

        let start = std::time::Instant::now();
        let res = RustRegistryUpstream::http_get_with(&agent, &format!("http://{addr}/whatever"));
        let elapsed = start.elapsed();

        assert!(res.is_err(), "a hung upstream must surface an error, not Ok");
        assert!(
            elapsed < Duration::from_secs(5),
            "GET must time out promptly (took {elapsed:?}), not block on the hung upstream"
        );
        let _ = handle.join();
    }

    /// LIVE network test — hits real crates.io. Ignored by default; run with
    /// `cargo test -p holger-server-lib --lib upstream_rust -- --ignored`.
    #[test]
    #[ignore]
    fn live_fetch_serde_from_crates_io() {
        let up = RustRegistryUpstream::crates_io("crates-io");
        let id = ArtifactId { namespace: None, name: "serde".into(), version: "1.0.0".into() };
        let got = up.fetch(&id).unwrap();
        assert!(got.is_some(), "crates.io must serve serde 1.0.0");
        assert!(got.unwrap().len() > 1000, "a real .crate is non-trivial in size");
    }
}