didbot_http/

lib.rs

//! One constructor for every outbound `reqwest` client this workspace
//! builds. A client with no timeout waits forever on a black-holed route.
//!
//! The workspace's `clippy.toml` disallows `reqwest`'s own constructors, so
//! a client built anywhere else fails the lint. A caller that needs another
//! total deadline, such as a held poll, sets it on the builder this returns;
//! the connect timeout stays.
//!
//! [`read_bounded`] is the other half: a `timeout` bounds how long a
//! response may take, not how much of it a caller will hold in memory.
//! `.json()` on an unbounded body is a memory sink against any server this
//! process does not run -- an operator's PDS, or whoever a DID document
//! points at.
//!
//! [`REFUSED_BLOCKS`] is the third: which addresses an outbound fetch may
//! reach at all. A URL named by a stranger, or by a document this server
//! just fetched, is aimed by somebody else; [`resolve_public`] and
//! [`pinned_client`] are how it stays aimed at the public internet.
//!
//! [`EXTRA_CA_CERTS`] is the fourth: which authorities a client here
//! trusts on top of the ones it is built with.
//!
//! [`doh`] is DNS over HTTPS, one GET through a pinned client, for the
//! records a stub resolver has no API for.
//!
//! [`lookup_host`] is how every client here resolves a name: RFC 6761's
//! `localhost` names go to loopback, and the rest go to the system.

mod address;
pub mod doh;
mod resolve;

pub use address::{
    is_loopback, is_public_address, pinned_client, refuse_private_host, resolve_public,
    AddressError, PrivateAddresses, REFUSED_BLOCKS,
};
pub use resolve::{is_localhost_name, lookup_host};

use std::time::Duration;

/// How long establishing a connection -- DNS, then the TCP and TLS
/// handshake -- may take, independent of [`DEFAULT_TIMEOUT`].
///
/// Set apart from the total deadline so a route that never completes a
/// handshake is a different fact, in a log line, from a server that
/// connects and then answers too slowly: the first is a network problem,
/// the second is the remote server's.
pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// The total deadline -- connect, send the request, and read the response --
/// for a request to a server this deployment does not run: an operator's
/// PDS, a relay, the ACME directory, a client's own metadata document.
///
/// Ten seconds, the value this workspace already used at its two busiest
/// outbound call sites (the relay announcement and the OAuth client
/// metadata fetch) before this crate existed. Long enough that an ordinary
/// server answers comfortably inside it; short enough that whatever is
/// waiting on the request -- a poll tick, a boot sequence -- finds out a
/// peer has stalled in single-digit seconds rather than never.
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);

/// The environment variable naming a PEM file of certificate authorities
/// every client built here trusts, on top of the ones it is built with.
///
/// This workspace's `reqwest` verifies against a bundled root store, which
/// means it ignores the machine's own — so a server serving a certificate
/// from `didbot_tls::local`, which is what a `.localhost` zone serves, is a
/// server these clients would otherwise refuse. Naming the authority's
/// certificate here is how a development stack is reached. The spelling
/// matches Node's `NODE_EXTRA_CA_CERTS`, which does the same job for the
/// Node tools beside these.
///
/// Additive, and only additive: nothing here can remove a root, and there
/// is no way to switch verification off.
pub const EXTRA_CA_CERTS: &str = "DIDBOT_EXTRA_CA_CERTS";

/// A [`reqwest::ClientBuilder`] pre-configured with
/// [`DEFAULT_CONNECT_TIMEOUT`] and [`DEFAULT_TIMEOUT`], for a caller that
/// needs to layer something else on top -- a redirect policy, a user agent
/// -- before building.
#[allow(clippy::disallowed_methods)]
pub fn builder() -> reqwest::ClientBuilder {
    let mut builder = reqwest::Client::builder()
        .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
        .timeout(DEFAULT_TIMEOUT)
        .dns_resolver(std::sync::Arc::new(resolve::Resolver));
    for certificate in extra_root_certificates() {
        builder = builder.add_root_certificate(certificate);
    }
    builder
}

/// Authorities a test added for this process; see
/// [`test_support::trust`]. Empty in anything that is not a test, and read
/// on every build so a test that adds one before its first request is
/// enough.
static PROCESS_ROOTS: std::sync::Mutex<Vec<reqwest::Certificate>> =
    std::sync::Mutex::new(Vec::new());

/// The authorities [`EXTRA_CA_CERTS`] names, or nothing if it is unset.
///
/// A file that cannot be read, or holds nothing a certificate parses out
/// of, contributes no roots and says so at `warn`. Failing the build
/// instead would take down a process over a development setting; failing
/// silently would leave every request refused with a handshake error that
/// names nothing.
fn extra_root_certificates() -> Vec<reqwest::Certificate> {
    let mut roots = PROCESS_ROOTS
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner())
        .clone();
    let Some(path) = std::env::var_os(EXTRA_CA_CERTS) else {
        return roots;
    };
    let path = std::path::PathBuf::from(path);
    let pem = match std::fs::read(&path) {
        Ok(pem) => pem,
        Err(error) => {
            tracing::warn!(
                %error,
                path = %path.display(),
                variable = EXTRA_CA_CERTS,
                "could not read the extra certificate authorities; trusting only the built-in roots"
            );
            return roots;
        }
    };
    match reqwest::Certificate::from_pem_bundle(&pem) {
        Ok(certificates) => roots.extend(certificates),
        Err(error) => tracing::warn!(
            %error,
            path = %path.display(),
            variable = EXTRA_CA_CERTS,
            "the extra certificate authorities did not parse; trusting only the built-in roots"
        ),
    }
    roots
}

/// [`builder`]'s two timeouts on a [`reqwest::blocking::ClientBuilder`], for
/// a caller with no async runtime to drive the async client on.
///
/// Build the client outside any async context: `reqwest`'s blocking client
/// panics when built inside one.
#[cfg(feature = "blocking")]
#[allow(clippy::disallowed_methods)]
pub fn blocking_builder() -> reqwest::blocking::ClientBuilder {
    let mut builder = reqwest::blocking::Client::builder()
        .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
        .timeout(DEFAULT_TIMEOUT)
        .dns_resolver(std::sync::Arc::new(resolve::Resolver));
    for certificate in extra_root_certificates() {
        builder = builder.add_root_certificate(certificate);
    }
    builder
}

/// The default outbound client: [`builder`], built.
///
/// Falls back to a default-configured [`reqwest::Client`] if construction
/// fails. Every caller of this function is on a path -- a boot sequence, a
/// poll loop -- where refusing to start at all is worse than running that
/// one process lifetime with `reqwest`'s own defaults instead of these.
///
/// The fallback says so at `warn`. Running a process lifetime on `reqwest`'s
/// defaults is the deliberate choice; doing it silently is what makes the
/// resulting hang -- a request with no deadline against a black-holed route
/// -- something nothing in the log explains.
pub fn client() -> reqwest::Client {
    builder().build().unwrap_or_else(|error| {
        tracing::warn!(
            %error,
            connect_timeout = ?DEFAULT_CONNECT_TIMEOUT,
            timeout = ?DEFAULT_TIMEOUT,
            "building the shared outbound HTTP client failed; falling back to reqwest's \
             own defaults, which set no deadline at all"
        );
        reqwest::Client::default()
    })
}

/// The most of one response body this workspace will hold in memory to read
/// it, when the response comes from a server nothing here operates.
///
/// Two mebibytes -- the same magnitude `didbot-serve`'s own
/// `MAX_REQUEST_BODY` already treats as a safe amount of one JSON payload
/// to hold in memory, on the inbound side of this same server. A DID
/// document or a page of an operator's own records has no business costing
/// more than one write this server would itself accept costs.
pub const MAX_JSON_BODY: usize = 2 * 1024 * 1024;

/// Why [`read_bounded`] did not return a body.
#[derive(Debug, thiserror::Error)]
pub enum ReadBoundedError {
    /// The body is larger than the bound it was read against -- either a
    /// declared `Content-Length` over the bound, refused before reading
    /// anything, or a body that crossed the bound while being read, which
    /// happens the moment it does rather than after it finishes.
    #[error("response body exceeds {max_bytes} bytes")]
    TooLarge {
        /// The bound the body exceeded.
        max_bytes: usize,
    },
    /// The connection or the HTTP exchange itself failed while reading.
    #[error("reading response body: {0}")]
    Transport(#[from] reqwest::Error),
    /// The same, for a blocking response, which reports through
    /// [`std::io`] rather than through `reqwest`.
    #[error("reading response body: {0}")]
    Io(#[from] std::io::Error),
}

/// Reads at most `max_bytes` of `response`'s body.
///
/// Neither path this can take ever holds more than one chunk past
/// `max_bytes` in memory: a declared `Content-Length` over the bound is
/// refused before reading anything, and a body that lies about its length,
/// or declines to declare one at all, is refused the instant it crosses the
/// bound rather than once it finishes.
pub async fn read_bounded(
    mut response: reqwest::Response,
    max_bytes: usize,
) -> Result<Vec<u8>, ReadBoundedError> {
    if response
        .content_length()
        .is_some_and(|len| len > max_bytes as u64)
    {
        return Err(ReadBoundedError::TooLarge { max_bytes });
    }
    let mut body = Vec::new();
    while let Some(chunk) = response.chunk().await? {
        if body.len() + chunk.len() > max_bytes {
            return Err(ReadBoundedError::TooLarge { max_bytes });
        }
        body.extend_from_slice(&chunk);
    }
    Ok(body)
}

/// [`read_bounded`], for a caller with no async runtime.
#[cfg(feature = "blocking")]
pub fn read_bounded_blocking(
    mut response: reqwest::blocking::Response,
    max_bytes: usize,
) -> Result<Vec<u8>, ReadBoundedError> {
    use std::io::Read;

    if response
        .content_length()
        .is_some_and(|len| len > max_bytes as u64)
    {
        return Err(ReadBoundedError::TooLarge { max_bytes });
    }
    // One byte past the bound, so a body that declares a small length and
    // then sends more is caught by what arrived rather than by what it said.
    let mut body = Vec::new();
    response
        .by_ref()
        .take(max_bytes as u64 + 1)
        .read_to_end(&mut body)?;
    if body.len() > max_bytes {
        return Err(ReadBoundedError::TooLarge { max_bytes });
    }
    Ok(body)
}

/// The body of a bounded read, as text.
///
/// Lossy, like `reqwest`'s own `text()`: a document that is not UTF-8 is
/// already going to fail whatever parses it, and saying so there names the
/// document rather than the encoding.
pub fn body_text(body: Vec<u8>) -> String {
    match String::from_utf8(body) {
        Ok(text) => text,
        Err(err) => String::from_utf8_lossy(err.as_bytes()).into_owned(),
    }
}

/// A server that has stopped answering, for tests of a client's timeouts.
#[cfg(feature = "test-support")]
pub mod test_support {
    use std::net::{SocketAddr, TcpListener};

    /// Trusts `ca_pem` in every client this crate builds for the rest of
    /// this process.
    ///
    /// [`super::EXTRA_CA_CERTS`] is how a running program is told, and a
    /// test cannot use it: a client built inside this workspace — a pinned
    /// one, or any caller's — is not a client a test can hand roots to, and
    /// an environment variable is shared by every test in the binary.
    ///
    /// Additive, like the variable, and only for a test: a server a test
    /// stood up signs from an authority it made, and this is how the code
    /// under test reaches it.
    pub fn trust(ca_pem: &str) {
        let certificate =
            reqwest::Certificate::from_pem(ca_pem.as_bytes()).expect("a test's own authority");
        super::PROCESS_ROOTS
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .push(certificate);
    }

    /// Binds a loopback listener that accepts every connection and never
    /// sends a byte, and returns its address. It lives until the process
    /// exits.
    ///
    /// It accepts on a thread, so it needs no runtime: a blocking client's
    /// test has none, and an async test may pause its clock, which fires a
    /// client's timeout the moment every task is idle.
    pub fn hung_server() -> SocketAddr {
        let listener = TcpListener::bind("127.0.0.1:0").expect("loopback binds");
        let addr = listener
            .local_addr()
            .expect("a bound listener has an address");
        std::thread::spawn(move || {
            let mut held = Vec::new();
            for stream in listener.incoming() {
                match stream {
                    Ok(stream) => held.push(stream),
                    Err(_) => break,
                }
            }
        });
        addr
    }
}

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

    fn response_with_body(body: Vec<u8>) -> reqwest::Response {
        http::Response::builder()
            .status(200)
            .body(body)
            .unwrap()
            .into()
    }

    #[tokio::test]
    async fn a_body_within_the_bound_is_read_whole() {
        let body = b"a modest json document".to_vec();
        let read = read_bounded(response_with_body(body.clone()), MAX_JSON_BODY)
            .await
            .unwrap();
        assert_eq!(read, body);
    }

    #[tokio::test]
    async fn a_declared_length_over_the_bound_is_refused() {
        let oversized = vec![b'x'; 1024];
        let err = read_bounded(response_with_body(oversized), 1023)
            .await
            .unwrap_err();
        assert!(matches!(
            err,
            ReadBoundedError::TooLarge { max_bytes: 1023 }
        ));
    }

    #[tokio::test]
    async fn a_body_exactly_at_the_bound_is_read_whole() {
        let exact = vec![b'x'; 1024];
        let read = read_bounded(response_with_body(exact.clone()), 1024)
            .await
            .unwrap();
        assert_eq!(read, exact);
    }

    /// A loopback server that answers every request with a chunked body it
    /// never finishes. Chunked means no `Content-Length`, so there is no
    /// header to refuse up front and the bound is the only thing stopping
    /// the read.
    #[cfg(feature = "blocking")]
    fn endless_chunked_server() -> std::net::SocketAddr {
        use std::io::{Read, Write};

        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("loopback binds");
        let addr = listener
            .local_addr()
            .expect("a bound listener has an address");
        std::thread::spawn(move || {
            for mut stream in listener.incoming().flatten() {
                // Read the request before answering it. A server that
                // writes its response while the client is still writing its
                // request gets the write reset under it, which the client
                // reports as a failed request rather than as the body this
                // test is about.
                let mut request = [0u8; 1024];
                if stream.read(&mut request).is_err() {
                    continue;
                }
                if stream
                    .write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n")
                    .is_err()
                {
                    continue;
                }
                let chunk = format!("2000\r\n{}\r\n", "x".repeat(0x2000));
                // Until the reader gives up and drops the connection.
                while stream.write_all(chunk.as_bytes()).is_ok() {}
            }
        });
        addr
    }

    #[cfg(feature = "blocking")]
    #[test]
    fn a_blocking_body_that_declares_no_length_is_still_cut_off_at_the_bound() {
        let addr = endless_chunked_server();
        let response = blocking_builder()
            .build()
            .expect("the blocking client builds")
            .get(format!("http://{addr}/"))
            .send()
            .expect("the server answers");
        assert!(response.content_length().is_none(), "chunked declares none");

        let err = read_bounded_blocking(response, 64 * 1024).unwrap_err();
        assert!(
            matches!(err, ReadBoundedError::TooLarge { max_bytes } if max_bytes == 64 * 1024),
            "{err:?}"
        );
    }

    /// A server that accepts the connection and never answers costs the
    /// default client exactly [`DEFAULT_TIMEOUT`], then a timeout error.
    #[tokio::test(start_paused = true)]
    async fn the_default_client_gives_up_on_a_server_that_never_answers() {
        let addr = test_support::hung_server();
        let started = tokio::time::Instant::now();
        let request = client().get(format!("http://{addr}/")).send();
        let err = tokio::time::timeout(DEFAULT_TIMEOUT * 2, request)
            .await
            .expect("the client's own timeout fires first")
            .expect_err("nothing ever answers");
        assert!(err.is_timeout(), "{err:?}");
        assert_eq!(started.elapsed(), DEFAULT_TIMEOUT);
    }

    /// The blocking client runs its own runtime, so this waits in real time.
    #[cfg(feature = "blocking")]
    #[test]
    fn the_blocking_client_gives_up_on_a_server_that_never_answers() {
        let addr = test_support::hung_server();
        let client = blocking_builder().build().expect("the client builds");
        let started = std::time::Instant::now();
        let err = client
            .get(format!("http://{addr}/"))
            .send()
            .expect_err("nothing ever answers");
        assert!(err.is_timeout(), "{err:?}");
        assert!(
            started.elapsed() >= DEFAULT_TIMEOUT,
            "{:?}",
            started.elapsed()
        );
    }
}