Something went wrong. Try again.
Identities for entities did.bot
agent llm did
Something went wrong. Try again.
18 kB · 453 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454//! 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.
mod address;pub mod doh;
pub use address::{ is_loopback, is_public_address, pinned_client, refuse_private_host, resolve_public, AddressError, PrivateAddresses, REFUSED_BLOCKS,};
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); 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); 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() ); }}