Something went wrong. Try again.
Chris’s PDS implementation
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386//! Stable error translation at the HTTP boundary.
use axum::{ Json, extract::Request, http::StatusCode, http::header::{ALLOW, CONTENT_TYPE, RETRY_AFTER}, middleware::Next, response::{IntoResponse, Response},};use serde::Serialize;
use crate::error::Error;
#[derive(Serialize)]pub(super) struct ErrorBody<Message> { pub(super) error: &'static str, pub(super) message: Message,}
impl IntoResponse for Error { fn into_response(self) -> Response { let (status, name) = match &self { Error::AuthenticationRequired => (StatusCode::UNAUTHORIZED, "AuthenticationRequired"), Error::Authentication { name, .. } => (StatusCode::UNAUTHORIZED, *name), Error::Authorization(_) => (StatusCode::FORBIDDEN, "Forbidden"), Error::NotFound => (StatusCode::NOT_FOUND, "RecordNotFound"), Error::Conflict(_) => (StatusCode::CONFLICT, "InvalidRequest"), Error::InvalidRequest(_) | Error::Url(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"), Error::RateLimited { .. } => (StatusCode::TOO_MANY_REQUESTS, "RateLimitExceeded"), Error::Xrpc { name, .. } if name.starts_with("Account") => { (StatusCode::FORBIDDEN, *name) } Error::Xrpc { name, .. } => (StatusCode::BAD_REQUEST, *name), Error::Upstream(_) => (StatusCode::BAD_GATEWAY, "UpstreamFailure"), Error::Configuration(_) | Error::Database(_) | Error::Io(_) | Error::Json(_) | Error::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError"), };
let message = if status.is_server_error() { tracing::error!(error = %self, "request failed"); match status { StatusCode::BAD_GATEWAY => "upstream service failed".to_owned(), _ => "internal server error".to_owned(), } } else { self.to_string() }; let mut response = response(status, name, message); if let Error::RateLimited { retry_after } = &self { response .headers_mut() .insert(RETRY_AFTER, (*retry_after).into()); } response }}
pub(super) fn response( status: StatusCode, name: &'static str, message: impl Into<String>,) -> Response { ( status, Json(ErrorBody { error: name, message: message.into(), }), ) .into_response()}
pub(super) async fn normalize_framework_rejections(request: Request, next: Next) -> Response { let is_xrpc = request.uri().path().starts_with("/xrpc/"); let response = next.run(request).await; if (!response.status().is_client_error() && !response.status().is_server_error()) || response .headers() .get(CONTENT_TYPE) .and_then(|value| value.to_str().ok()) .is_some_and(|value| value.starts_with("application/json")) { return response; }
let original_status = response.status(); let (status, name, message) = match original_status { StatusCode::NOT_FOUND if is_xrpc => ( StatusCode::NOT_FOUND, "MethodNotFound", "XRPC method is not implemented", ), StatusCode::NOT_FOUND => ( StatusCode::NOT_FOUND, "NotFound", "route is not implemented", ), StatusCode::METHOD_NOT_ALLOWED => ( StatusCode::METHOD_NOT_ALLOWED, "MethodNotAllowed", "XRPC method does not support this HTTP verb", ), StatusCode::PAYLOAD_TOO_LARGE => ( StatusCode::PAYLOAD_TOO_LARGE, "PayloadTooLarge", "request body exceeds the endpoint limit", ), status if status.is_client_error() => ( StatusCode::BAD_REQUEST, "InvalidRequest", "request does not match the endpoint schema", ), _ => ( StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError", "request framework failed", ), }; let allow = response.headers().get(ALLOW).cloned(); let mut normalized = self::response(status, name, message); if let Some(allow) = allow { normalized.headers_mut().insert(ALLOW, allow); } normalized}
#[cfg(test)]mod tests { use axum::{ Router, body::{Body, to_bytes}, http::HeaderValue, middleware::from_fn, }; use rand::{Rng as _, SeedableRng as _, rngs::StdRng}; use tower::ServiceExt as _;
use super::*;
#[tokio::test] async fn shared_error_body_keeps_owned_json_response_bytes() { #[derive(Serialize)] struct PreviousErrorBody { error: &'static str, message: String, } let messages = [ String::new(), "cursor is in the future".to_owned(), "quote-\"-backslash-\\-雪-🦀".to_owned(), (0u8..=31).map(char::from).collect(), "a".repeat(4096), ]; for name in ["", "FutureCursor", "Error-\"\\\n-雪"] { for message in &messages { for status in [ StatusCode::OK, StatusCode::BAD_REQUEST, StatusCode::BAD_GATEWAY, ] { let previous = ( status, Json(PreviousErrorBody { error: name, message: message.clone(), }), ) .into_response(); let actual = response(status, name, message.clone()); assert_eq!(actual.status(), previous.status()); assert_eq!(actual.headers(), previous.headers()); assert_eq!( to_bytes(actual.into_body(), usize::MAX).await.unwrap(), to_bytes(previous.into_body(), usize::MAX).await.unwrap(), ); } } } }
#[tokio::test] async fn security_headers_preserve_exact_paths_statuses_and_header_replacement() { use axum::http::{HeaderMap, header::CONTENT_LENGTH};
for (path, blob, cacheable) in [ ("/xrpc/com.atproto.sync.getBlob", true, true), ("/xrpc/com.atproto.sync.getBlob?ignored=1", true, true), ("/xrpc/com.atproto.space.getBlob", true, false), ("/xrpc/com.atproto.space.getBlob?ignored=1", true, false), ("/xrpc/com.atproto.sync.getBlob/", false, false), ("/xrpc/com.atproto.sync.getBlobOther", false, false), ("/xrpc/com.atproto.sync.getblob", false, false), ("/xrpc/com.atproto.sync.get%42lob", false, false), ("/.well-known/did.json", false, true), ("/.well-known", false, false), ("/oauth/token", false, false), ] { for status in [200, 206, 302, 400, 500].map(|status| StatusCode::from_u16(status).unwrap()) { for preexisting in [false, true] { let policy = [ ("x-content-type-options", "nosniff"), ("referrer-policy", "no-referrer"), ("x-frame-options", "DENY"), ( "content-security-policy", if blob { "default-src 'none'; sandbox" } else { "default-src 'none'; base-uri 'none'; form-action 'self'; frame-ancestors 'none'" }, ), ( "cache-control", if cacheable && status.is_success() { "public, max-age=300" } else { "no-store" }, ), ]; let mut original = HeaderMap::new(); original.insert(CONTENT_TYPE, HeaderValue::from_static("text/plain")); original.insert(CONTENT_LENGTH, HeaderValue::from_static("13")); original.insert("x-original", HeaderValue::from_static("untouched")); if preexisting { for (name, _) in policy { let mut old = HeaderValue::from_static("previous policy"); old.set_sensitive(true); original.append(name, old.clone()); original.append(name, old); } } let mut expected = original.clone(); for (name, value) in policy { expected.insert(name, HeaderValue::from_static(value)); } let app = Router::new() .fallback(move || { let headers = original.clone(); async move { let mut response = Response::new(Body::from("original body")); *response.status_mut() = status; *response.headers_mut() = headers; response } }) .layer(from_fn(crate::http::security::response_headers)); let response = app .oneshot(axum::http::Request::get(path).body(Body::empty()).unwrap()) .await .unwrap(); assert_eq!(response.status(), status, "{path}"); assert_eq!(response.headers(), &expected, "{path}: {status}"); for (name, _) in policy { assert_eq!(response.headers().get_all(name).iter().count(), 1); assert!(!response.headers()[name].is_sensitive()); } assert_eq!( to_bytes(response.into_body(), usize::MAX) .await .unwrap() .as_ref(), b"original body", ); } } } }
#[tokio::test] async fn framework_normalization_preserves_exact_content_type_prefix_semantics() { let cases: &[(&[&[u8]], bool)] = &[ (&[], false), (&[b"application/json"], true), (&[b"application/json; charset=utf-8"], true), (&[b"application/json-not-a-mime-type"], true), (&[b"Application/JSON"], false), (&[b" application/json"], false), (&[b"text/plain"], false), (&[b"application/json\x80"], false), (&[b"application/json", b"text/plain"], true), (&[b"text/plain", b"application/json"], false), ]; for (content_types, is_json) in cases { for status in [ StatusCode::OK, StatusCode::BAD_REQUEST, StatusCode::INTERNAL_SERVER_ERROR, ] { let mut headers = axum::http::HeaderMap::new(); headers.insert(ALLOW, HeaderValue::from_static("GET,HEAD")); headers.insert("x-original", HeaderValue::from_static("kept")); headers.insert( axum::http::header::CONTENT_LENGTH, HeaderValue::from("original body\0not JSON".len()), ); for content_type in *content_types { headers.append(CONTENT_TYPE, HeaderValue::from_bytes(content_type).unwrap()); } let expected_headers = headers.clone(); let app = Router::new() .fallback(move || { let headers = headers.clone(); async move { let mut response = Response::new(Body::from("original body\0not JSON")); *response.status_mut() = status; *response.headers_mut() = headers; response } }) .layer(from_fn(normalize_framework_rejections)); let actual = app .oneshot( axum::http::Request::get("/xrpc/test") .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(actual.status(), status); assert_eq!(actual.headers()[ALLOW], expected_headers[ALLOW]); let preserved = status == StatusCode::OK || *is_json; if preserved { assert_eq!(actual.headers(), &expected_headers); } else { assert_eq!(actual.headers()[CONTENT_TYPE], "application/json"); assert!(!actual.headers().contains_key("x-original")); } let body = to_bytes(actual.into_body(), usize::MAX).await.unwrap(); assert_eq!(body.as_ref() == b"original body\0not JSON", preserved); } } }
#[tokio::test] async fn native_retry_after_preserves_header_and_complete_error_response() { let mut values = vec![0, u64::MAX]; for exponent in 0..20 { let boundary = 10_u64.pow(exponent); values.extend([boundary - 1, boundary, boundary + 1]); } let mut random = StdRng::seed_from_u64(0x5f4d_893c_127a_6e20); values.extend((0..1024).map(|_| random.random::<u64>()));
for retry_after in values { let previous: HeaderValue = retry_after .to_string() .parse() .expect("integer Retry-After is valid"); let native = HeaderValue::from(retry_after); assert_eq!(native.as_bytes(), previous.as_bytes()); assert_eq!(native.to_str().unwrap(), retry_after.to_string()); assert_eq!(native.is_sensitive(), previous.is_sensitive()); assert!(!native.is_sensitive()); assert_eq!(format!("{native:?}"), format!("{previous:?}"));
let mut expected = response( StatusCode::TOO_MANY_REQUESTS, "RateLimitExceeded", "authenticated principal request rate exceeded", ); expected.headers_mut().insert(RETRY_AFTER, previous); let actual = Error::RateLimited { retry_after }.into_response(); assert_eq!(actual.status(), expected.status()); assert_eq!(actual.version(), expected.version()); assert_eq!(actual.headers(), expected.headers()); assert!(!actual.headers()[RETRY_AFTER].is_sensitive()); let expected = to_bytes(expected.into_body(), usize::MAX).await.unwrap(); assert_eq!( expected.as_ref(), br#"{"error":"RateLimitExceeded","message":"authenticated principal request rate exceeded"}"#, ); assert_eq!( to_bytes(actual.into_body(), usize::MAX).await.unwrap(), expected, ); } }}