diff --git a/crates/didbot-serve/src/lib.rs b/crates/didbot-serve/src/lib.rs index c7312ac9..dd491122 100644 --- a/crates/didbot-serve/src/lib.rs +++ b/crates/didbot-serve/src/lib.rs @@ -96,7 +96,7 @@ pub use oauth::scope::{ pub use ownership_poll::{OwnershipPoll, PollNudge, PollSchedule, MIN_POLL_INTERVAL}; pub use routes::{ app, app_with_auth, app_with_events, app_with_health, app_with_repos, app_with_streams, - AuthState, ATPROTO_METHODS, BOT_DID_METHODS, + AuthState, XrpcBody, ATPROTO_METHODS, BOT_DID_METHODS, MAX_REQUEST_BODY, }; pub use routes::{TURN_HEADER, TURN_HEADER_MAX_LENGTH}; pub use subscribe::{ diff --git a/crates/didbot-serve/src/routes.rs b/crates/didbot-serve/src/routes.rs index 2b3e31b0..eb78dc80 100644 --- a/crates/didbot-serve/src/routes.rs +++ b/crates/didbot-serve/src/routes.rs @@ -555,6 +555,10 @@ pub fn app_with_repos( StatusCode::SERVICE_UNAVAILABLE, REQUEST_TIMEOUT, )) + // The buffering bound, pinned rather than inherited; see + // `MAX_REQUEST_BODY`. `uploadBlob` is unaffected because it takes the + // body as a stream, which this layer does not reach. + .layer(axum::extract::DefaultBodyLimit::max(MAX_REQUEST_BODY)) } /// Parses a JSON body, mapping any failure to a 400 in the atproto shape. @@ -565,6 +569,66 @@ fn parse_body(body: &Bytes) -> Result { serde_json::from_slice(body).map_err(|err| ApiError::bad_request(err.to_string())) } +/// The largest request body this server will hold in memory. +/// +/// Two mebibytes, which is what axum bounds a buffering extractor to when +/// nothing says otherwise. It is written down here rather than inherited +/// silently, because it is the only thing standing between an anonymous +/// caller and this process's memory, and a bound that arrives from a +/// dependency's default is one nothing in this repository would notice +/// changing. +/// +/// It is not a bound on a *blob*: `com.atproto.repo.uploadBlob` takes the +/// body as a stream and never buffers it, and its size limit is the blob +/// store's own. This is the bound on the JSON every other procedure sends. +/// +/// Enforcement is streaming rather than after the fact — the body is refused +/// as it arrives, at the first frame that carries the request past the bound, +/// so a caller announcing or sending a gigabyte never has a gigabyte of it +/// resident here. +pub const MAX_REQUEST_BODY: usize = 2 * 1024 * 1024; + +/// A buffered request body, refused in the atproto shape. +/// +/// `Bytes` on its own is an extractor whose rejection is axum's own +/// plain-text sentence with no `error` name at all — the same reason +/// `parse_body` exists rather than `axum::Json`, and the same reason every +/// query below is extracted as a `Result`. A body over +/// [`MAX_REQUEST_BODY`] is the one failure on this surface a caller cannot +/// avoid by writing a better request, so it is the one that most has to be +/// machine-readable. +pub struct XrpcBody(pub Bytes); + +impl axum::extract::FromRequest for XrpcBody +where + S: Send + Sync, +{ + type Rejection = ApiError; + + async fn from_request( + request: axum::extract::Request, + state: &S, + ) -> Result { + match Bytes::from_request(request, state).await { + Ok(body) => Ok(Self(body)), + // `RequestTooLarge` names no error the lexicons declare, the + // same position `RateLimitExceeded` and `ServerNotReady` are in: + // 413 is the standard status and every client already knows it, + // and the name is there for the ones that switch on it. The + // message is this crate's own rather than the rejection's, so + // nothing about how the body was buffered reaches the caller. + Err(rejection) if rejection.status() == StatusCode::PAYLOAD_TOO_LARGE => { + Err(ApiError::new( + StatusCode::PAYLOAD_TOO_LARGE, + "RequestTooLarge", + format!("a request body may not exceed {MAX_REQUEST_BODY} bytes"), + )) + } + Err(rejection) => Err(ApiError::bad_request(rejection.body_text())), + } + } +} + /// Resolves a repo write route's credential — an agent token or a DPoP-bound /// OAuth access token. See `auth::require_agent_token_or_dpop`. fn resource_credential( @@ -1090,7 +1154,7 @@ async fn firehose_stream( /// of which close it to every caller at once rather than to one; see also /// `plan/auth-types.md`'s note that the operator surface is kept separate /// rather than folded into every admin-shaped route. -async fn provision_agent(State(state): State, body: Bytes) -> Response { +async fn provision_agent(State(state): State, XrpcBody(body): XrpcBody) -> Response { // Both of these run before the attestation claim is read, because a // server that will not provision has no business spending one. // @@ -1150,7 +1214,11 @@ async fn provision_agent(State(state): State, body: Bytes) -> Response /// credential — the self-service case [`auth::require_self`] exists for, /// and the only case, since this server holds no credential that could /// authenticate somebody acting on another account. -async fn delete_agent(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn delete_agent( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { // `docs/write-pipeline.md`'s stage 1, ahead of everything else this // route does. A `Revoke` stops outstanding work, and an agent acting on // its own account with a credential issued before the stop is @@ -1188,7 +1256,11 @@ async fn delete_agent(State(state): State, headers: HeaderMap, body: B /// it under a stop would leave a caller less able to restrain itself than /// before the operator pulled the brake. A gate that can only deny is safe /// to skip for an operation that can only deny. -async fn freeze_agent(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn freeze_agent( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { let request: DidRequest = match parse_body(&body) { Ok(request) => request, Err(err) => return err.into_response(), @@ -1209,7 +1281,7 @@ async fn freeze_agent(State(state): State, headers: HeaderMap, body: B async fn unfreeze_agent( State(state): State, headers: HeaderMap, - body: Bytes, + XrpcBody(body): XrpcBody, ) -> Response { // `docs/write-pipeline.md`'s stage 1, ahead of everything else this // route does. A `Revoke` stops outstanding work, and an agent acting on @@ -1247,7 +1319,7 @@ async fn unfreeze_agent( async fn soft_delete_agent( State(state): State, headers: HeaderMap, - body: Bytes, + XrpcBody(body): XrpcBody, ) -> Response { // `docs/write-pipeline.md`'s stage 1, ahead of everything else this // route does. A `Revoke` stops outstanding work, and an agent acting on @@ -1277,7 +1349,11 @@ async fn soft_delete_agent( /// /// `Credential::AgentSelf`, the same as [`delete_agent`]: an agent may pin /// or unpin itself, and nothing else may pin it over HTTP. -async fn set_pinned(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn set_pinned( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { // `docs/write-pipeline.md`'s stage 1, ahead of everything else this // route does. A `Revoke` stops outstanding work, and an agent acting on // its own account with a credential issued before the stop is @@ -1440,7 +1516,11 @@ async fn list_ledgers(State(state): State) -> Response { /// declares, and for a reason about names: `InvalidSwap` tells a /// client to re-read and retry, and this retry can never succeed. A caller /// that meant to replace the record wants `putRecord`. -async fn create_record(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn create_record( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads // the body or spends a credential: a halted server and a server not // accepting writes are both public facts -- `/onboarding` publishes the @@ -1496,7 +1576,11 @@ async fn create_record(State(state): State, headers: HeaderMap, body: /// replaced rather than refused, and that is now the difference between this /// and [`create_record`]. Both `swapRecord` and `swapCommit` are honoured, /// and `swapRecord: null` asserts the record does not exist yet. -async fn put_record(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn put_record( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads // the body or spends a credential: a halted server and a server not // accepting writes are both public facts -- `/onboarding` publishes the @@ -1746,7 +1830,11 @@ pub(crate) fn require_record_writes(state: &AppState) -> Result<(), ApiError> { /// /// The response omits `commit`, which the lexicon marks optional. There is no /// commit here to report. -async fn delete_record(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn delete_record( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads // the body or spends a credential: a halted server and a server not // accepting writes are both public facts -- `/onboarding` publishes the @@ -1808,7 +1896,11 @@ async fn delete_record(State(state): State, headers: HeaderMap, body: /// compare-and-swap against the commit this batch is about to replace, not a /// look before it: it is checked under the same lock the batch's one commit /// is built under. -async fn apply_writes(State(state): State, headers: HeaderMap, body: Bytes) -> Response { +async fn apply_writes( + State(state): State, + headers: HeaderMap, + XrpcBody(body): XrpcBody, +) -> Response { // Stages 1 and 2 of `docs/write-pipeline.md`, before this route reads // the body or spends a credential: a halted server and a server not // accepting writes are both public facts -- `/onboarding` publishes the @@ -2619,7 +2711,7 @@ async fn create_session( State(state): State, peer: Peer, headers: HeaderMap, - body: Bytes, + XrpcBody(body): XrpcBody, ) -> Response { if let Err(halted) = state.estop.check_issue(EstopRefusal::Token) { return ApiError::from(halted).into_response(); diff --git a/crates/didbot-serve/tests/xrpc_envelope.rs b/crates/didbot-serve/tests/xrpc_envelope.rs index de8dc37d..7d70fd91 100644 --- a/crates/didbot-serve/tests/xrpc_envelope.rs +++ b/crates/didbot-serve/tests/xrpc_envelope.rs @@ -126,3 +126,68 @@ async fn a_query_key_running_into_a_multi_byte_character_is_answered() { let (error, _) = error_body(response).await; assert_eq!(error, "InvalidRequest"); } + +/// A body past what this server will buffer is refused in the atproto shape, +/// and refused before any credential is looked at. +/// +/// The buffering bound is real either way — it is enforced as the body +/// arrives, so nothing here is ever resident — but until this was wired the +/// refusal came out of the extractor as `text/plain`, which is a failure no +/// client can match on. +#[tokio::test] +async fn a_body_past_the_buffering_limit_is_refused_in_the_atproto_shape() { + let (app, _) = router(); + let oversized = vec![b'x'; didbot_serve::MAX_REQUEST_BODY + 1]; + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/xrpc/com.atproto.repo.createRecord") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(oversized)) + .expect("request builds"), + ) + .await + .expect("router is infallible"); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + let (error, message) = error_body(response).await; + assert_eq!(error, "RequestTooLarge"); + assert!( + !message.contains("::") && !message.contains(".rs"), + "a message is a sentence for a human, not a Rust path: {message}" + ); +} + +/// A body just under the bound is not refused by it. +/// +/// The pair to the test above: a limit that refused everything would pass +/// that one for the wrong reason. `AuthenticationRequired` is as far as this +/// gets, which is exactly the point — the body was buffered and the request +/// reached the credential check. +#[tokio::test] +async fn a_body_under_the_buffering_limit_reaches_the_credential_check() { + let (app, did) = router(); + let mut record = serde_json::json!({ + "repo": did, + "collection": "app.bsky.feed.post", + "record": { "$type": "app.bsky.feed.post", "text": "" }, + }); + let padding = "x".repeat(didbot_serve::MAX_REQUEST_BODY / 2); + record["record"]["text"] = Value::String(padding); + let body = serde_json::to_vec(&record).expect("body serializes"); + assert!(body.len() < didbot_serve::MAX_REQUEST_BODY); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/xrpc/com.atproto.repo.createRecord") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .expect("request builds"), + ) + .await + .expect("router is infallible"); + assert_ne!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + let (error, _) = error_body(response).await; + assert_eq!(error, "AuthenticationRequired"); +}