Something went wrong. Try again.
atproto git client
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332//! The loopback HTTP plumbing the four mocks sit on.//!//! One listener per service on a port the kernel picks, so nothing here//! collides with another test, another suite running in parallel, or//! anything else on the machine. Bound before the base URL is handed out, so//! a command can never race the listener into existence.//!//! hyper's server half rather than a hand-rolled socket reader: reqwest —//! which is what atgc uses for every request — is built on hyper's client//! half, and the parts of HTTP/1.1 that a hand-rolled server gets wrong are//! exactly the parts a real client exercises without asking (keep-alive,//! chunked bodies, `Expect: 100-continue`). It is already in the tree.
use std::convert::Infallible;use std::net::SocketAddr;use std::pin::Pin;use std::sync::{Arc, Mutex};use std::task::{Context, Poll};
use http_body_util::combinators::BoxBody;use http_body_util::{BodyExt, Full};use hyper::body::{Bytes, Frame};use hyper::service::service_fn;use hyper::{Request, Response, StatusCode};use hyper_util::rt::TokioIo;
use super::world::World;
/// Every reply is boxed so one of them can be a stream.////// `Full<Bytes>` knows its own length, and hyper turns that into a/// `Content-Length` header. That is right for the four ordinary replies and/// useless for [`Reply::Flood`], whose entire point is to arrive without/// one — so the body type is erased and hyper decides per response.type Body = BoxBody<Bytes, Infallible>;
/// A request, flattened into the parts a mock cares about.pub struct Incoming { pub method: hyper::Method, /// Path with no leading slash. pub path: String, pub params: std::collections::BTreeMap<String, String>, pub body: Vec<u8>, /// The `Authorization` header's credential, with the scheme removed. pub token: Option<String>,}
impl Incoming { pub fn param(&self, name: &str) -> Option<&str> { self.params.get(name).map(String::as_str) }
/// The request body as JSON, or `Null` when there is none. pub fn json(&self) -> serde_json::Value { serde_json::from_slice(&self.body).unwrap_or(serde_json::Value::Null) }
/// The XRPC method this path names, if it is an XRPC path at all. pub fn nsid(&self) -> Option<&str> { self.path.strip_prefix("xrpc/") }}
/// What a mock answers with.pub enum Reply { Json(serde_json::Value), /// JSON under a status other than 200. RFC 9126 has a pushed /// authorization request answer `201 Created`, and jacquard checks. JsonStatus(StatusCode, serde_json::Value), Bytes(Vec<u8>), /// An XRPC error: status, `error` name, and a message. Error(StatusCode, &'static str, String), /// This many bytes, chunked, with no `Content-Length` — a host that /// answers with more than atgc will read. See [`World::flood`]. Flood(usize), /// A `302` to `location`. The appview publishes the repo-DID-to-owner /// mapping this way and nowhere else, so a mock that cannot redirect /// cannot answer the one question `resolve::owner_of` asks. Redirect(String),}
impl Reply { pub fn ok(value: serde_json::Value) -> Reply { Reply::Json(value) }
pub fn not_found(message: impl Into<String>) -> Reply { Reply::Error(StatusCode::NOT_FOUND, "RecordNotFound", message.into()) }
pub fn bad(error: &'static str, message: impl Into<String>) -> Reply { Reply::Error(StatusCode::BAD_REQUEST, error, message.into()) }
/// A knot turning down a call it understood, in the knot's own spelling: /// 401, tagged `AccessControl`. pub fn denied(message: impl Into<String>) -> Reply { Reply::Error(StatusCode::UNAUTHORIZED, "AccessControl", message.into()) }
/// A knot refusing because the thing is already there — the 409 that /// `sh.tangled.repo.create` answers a taken name with. Its own /// constructor because `exit::from_status` classifies it apart from every /// other refusal, and a test that could not produce one could not check /// that classification. pub fn conflict(message: impl Into<String>) -> Reply { Reply::Error(StatusCode::CONFLICT, "RepoExists", message.into()) }}
/// A bound service: the base URL to point atgc at, and the task serving it.pub struct Service { pub base: String,}
/// The one runtime every mock in this test binary runs on.////// Shared rather than one per scenario, and that is not a micro-optimization:/// cargo runs integration tests on one thread per test, so a runtime per/// scenario meant dozens of tokio worker threads competing with dozens of/// `atgc` child processes on the same machine. Under that load a loopback/// request could go unanswered long enough for a command to give up, and the/// suite failed in a way that looked like a bug in the tool — a DID that/// would not resolve — rather than like the harness running out of room.////// Nothing is lost by sharing it. The listeners are independent, the state/// behind them is per-scenario, and it never shuts down because the process/// exits when the last test does.fn runtime() -> &'static tokio::runtime::Runtime { static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new(); RUNTIME.get_or_init(|| { tokio::runtime::Builder::new_multi_thread() .worker_threads(2) .enable_all() .build() .expect("a runtime for the mocks") })}
/// Bind a listener and serve `handler` on it for the rest of the process.////// `handler` is synchronous and takes the world's lock for the length of one/// request, which serializes one scenario's mocks. That is the right trade/// here: no command atgc runs depends on two of its own requests being in/// flight at once, and a journal written under one lock is a journal whose/// order is the order the requests actually arrived in.pub fn serve<F>(world: Arc<Mutex<World>>, handler: F) -> std::io::Result<Service>where F: Fn(&mut World, &Incoming) -> Reply + Send + Sync + 'static,{ let rt = runtime(); let handler = Arc::new(handler); // Bound synchronously so the port is real before this returns. let listener = std::net::TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))?; let base = format!("http://{}", listener.local_addr()?); listener.set_nonblocking(true)?;
let _guard = rt.enter(); let listener = tokio::net::TcpListener::from_std(listener)?; rt.spawn(async move { loop { // A failed accept is not a reason to stop listening. This used // to `return`, which turns one transient error into a service // that is gone for the rest of the scenario — and its absence // then surfaces as whatever the command downstream says when a // DID will not resolve, which is a long way from the cause. let Ok((stream, _)) = listener.accept().await else { tokio::task::yield_now().await; continue; }; let world = world.clone(); let handler = handler.clone(); tokio::spawn(async move { let service = service_fn(move |req: Request<hyper::body::Incoming>| { let world = world.clone(); let handler = handler.clone(); async move { Ok::<_, Infallible>(answer(&world, handler.as_ref(), req).await) } }); let _ = hyper::server::conn::http1::Builder::new() .serve_connection(TokioIo::new(stream), service) .await; }); } }); Ok(Service { base })}
async fn answer<F>( world: &Mutex<World>, handler: &F, req: Request<hyper::body::Incoming>,) -> Response<Body>where F: Fn(&mut World, &Incoming) -> Reply,{ let method = req.method().clone(); let token = req .headers() .get(hyper::header::AUTHORIZATION) .and_then(|v| v.to_str().ok()) // "DPoP <token>" and "Bearer <token>" both reduce to the credential. .map(|v| v.split_whitespace().last().unwrap_or(v).to_string()); let uri = req.uri().clone(); let body = req .into_body() .collect() .await .map(|b| b.to_bytes().to_vec()) .unwrap_or_default();
let incoming = Incoming { method, // The path arrives percent-encoded — a DID's colons survive, but a // strict client may encode them, and `at://` URIs in query strings // certainly are. path: decode(uri.path().trim_start_matches('/')), params: query(uri.query().unwrap_or_default()), body, token, };
let reply = { let mut world = world.lock().expect("the world's lock"); handler(&mut world, &incoming) };
match reply { Reply::JsonStatus(status, value) => Response::builder() .status(status) .header(hyper::header::CONTENT_TYPE, "application/json") .body(Full::new(Bytes::from(value.to_string())).boxed()) .expect("a response"), Reply::Json(value) => Response::builder() .status(StatusCode::OK) .header(hyper::header::CONTENT_TYPE, "application/json") .body(Full::new(Bytes::from(value.to_string())).boxed()) .expect("a response"), Reply::Bytes(bytes) => Response::builder() .status(StatusCode::OK) .header(hyper::header::CONTENT_TYPE, "application/octet-stream") .body(Full::new(Bytes::from(bytes)).boxed()) .expect("a response"), Reply::Error(status, error, message) => { let body = serde_json::json!({ "error": error, "message": message }); Response::builder() .status(status) .header(hyper::header::CONTENT_TYPE, "application/json") .body(Full::new(Bytes::from(body.to_string())).boxed()) .expect("a response") } Reply::Redirect(location) => Response::builder() .status(StatusCode::FOUND) .header(hyper::header::LOCATION, location) .body(Full::new(Bytes::new()).boxed()) .expect("a response"), // A 200 with a plausible content type, because the interesting // refusal is the one that happens before anything has been parsed // or the status has been looked at. Reply::Flood(bytes) => Response::builder() .status(StatusCode::OK) .header(hyper::header::CONTENT_TYPE, "application/json") .body(Flood { remaining: bytes }.boxed()) .expect("a response"), }}
/// A body of `remaining` bytes that never says how many there will be.////// The shape the size bound actually has to survive. Everything built on/// `Full<Bytes>` carries a `Content-Length`, so a mock made of those/// exercises only the cheap half of `http::bytes_bounded` — the half a/// sender can omit for free by answering chunked, which is what this does./// What is left holding the line is the running total, which is the thing/// worth testing end to end.////// It never yields `Poll::Pending`, so there is no waker to store: a real/// hostile server would trickle, and trickling is precisely what/// `READ_TIMEOUT` already covers. What it cannot cover is a server that/// answers *promptly* and forever, and that is this.struct Flood { remaining: usize,}
impl hyper::body::Body for Flood { type Data = Bytes; type Error = Infallible;
fn poll_frame( mut self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> { if self.remaining == 0 { return Poll::Ready(None); } // 64 KiB a frame: few enough frames that nine mebibytes is fast, and // small enough that the running total is consulted a hundred-odd // times on the way to the limit rather than sailing past it in one. let n = self.remaining.min(64 * 1024); self.remaining -= n; Poll::Ready(Some(Ok(Frame::data(Bytes::from(vec![b'x'; n]))))) }}
fn query(raw: &str) -> std::collections::BTreeMap<String, String> { raw.split('&') .filter(|pair| !pair.is_empty()) .filter_map(|pair| pair.split_once('=')) .map(|(k, v)| (decode(k), decode(v))) .collect()}
/// Percent-decoding, enough for the values atgc puts in a query: DIDs,/// NSIDs, record keys, CIDs and `at://` URIs.fn decode(raw: &str) -> String { let bytes = raw.as_bytes(); let mut out = Vec::with_capacity(bytes.len()); let mut i = 0; while i < bytes.len() { if bytes[i] == b'%' && i + 2 < bytes.len() { let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).unwrap_or(""); if let Ok(byte) = u8::from_str_radix(hex, 16) { out.push(byte); i += 3; continue; } } out.push(bytes[i]); i += 1; } String::from_utf8_lossy(&out).into_owned()}