Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
8.6 kB · 247 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248pub mod api;pub mod auth;pub mod cache;pub mod cache_keys;pub mod cid_types;pub mod circuit_breaker;pub mod comms;pub mod config;pub mod crawlers;pub mod delegation;pub mod did;pub mod handle;pub mod image;pub mod metrics;pub mod moderation;pub mod oauth;pub mod plc;pub mod rate_limit;pub mod repo;pub mod repo_ops;pub mod repo_write_lock;pub mod scheduled;pub mod sso;pub mod state;pub mod storage;pub mod sync;pub mod types;pub mod util;pub mod validation;
use api::proxy::XrpcProxyLayer;use axum::{Json, Router, extract::DefaultBodyLimit, http::Method, middleware, routing::get};use http::StatusCode;use serde_json::json;use state::AppState;use tower::ServiceBuilder;use tower_http::{ cors::{Any, CorsLayer}, services::{ServeDir, ServeFile},};pub use tranquil_db_traits::AccountStatus;pub use types::{AccountState, AtIdentifier, AtUri, Did, Handle, Nsid, Rkey};
#[cfg(debug_assertions)]pub const BUILD_VERSION: &str = concat!( env!("CARGO_PKG_VERSION"), " (built ", env!("BUILD_TIMESTAMP"), ")");#[cfg(not(debug_assertions))]pub const BUILD_VERSION: &str = env!("CARGO_PKG_VERSION");
pub const GENERAL_BODY_LIMIT: usize = 16 * 1024 * 1024;
pub struct ExternalRoutes { pub xrpc: Router<AppState>, pub oauth: Router<AppState>, pub well_known: Router<AppState>, pub extra: Router<AppState>,}
impl Default for ExternalRoutes { fn default() -> Self { Self { xrpc: Router::new(), oauth: Router::new(), well_known: Router::new(), extra: Router::new(), } }}
pub fn app(state: AppState) -> Router { app_with_routes(state, ExternalRoutes::default())}
pub fn app_with_routes(state: AppState, external: ExternalRoutes) -> Router { let xrpc_router = external.xrpc .fallback(async || ( StatusCode::NOT_IMPLEMENTED, Json(json!({"error": "MethodNotImplemented", "message": "Method not implemented. For app.bsky.* methods, include an atproto-proxy header specifying your AppView."})), )); let xrpc_service = ServiceBuilder::new() .layer(XrpcProxyLayer::new(state.clone())) .service( xrpc_router .layer(middleware::from_fn(oauth::verify::dpop_nonce_middleware)) .with_state(state.clone()), );
let oauth_router = external.oauth;
let well_known_router = external.well_known;
let router = Router::new() .nest_service("/xrpc", xrpc_service) .nest("/oauth", oauth_router) .nest("/.well-known", well_known_router) .route("/metrics", get(metrics::metrics_handler)) .merge(external.extra) .layer(middleware::from_fn_with_state( state.clone(), rate_limit::global_xrpc_rate_limit, )) .layer(DefaultBodyLimit::max(GENERAL_BODY_LIMIT)) .layer(axum::middleware::map_response(rewrite_extractor_errors)) .layer(middleware::from_fn(metrics::metrics_middleware)) .layer( CorsLayer::new() .allow_origin(Any) .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) .allow_headers([ http::header::AUTHORIZATION, http::header::CONTENT_TYPE, http::header::CONTENT_ENCODING, http::header::ACCEPT_ENCODING, http::header::USER_AGENT, util::HEADER_DPOP, util::HEADER_ATPROTO_PROXY, util::HEADER_ATPROTO_ACCEPT_LABELERS, util::HEADER_X_BSKY_TOPICS, ]) .expose_headers([ http::header::WWW_AUTHENTICATE, util::HEADER_DPOP_NONCE, util::HEADER_ATPROTO_REPO_REV, util::HEADER_ATPROTO_CONTENT_LABELERS, http::HeaderName::from_static("ratelimit-limit"), http::HeaderName::from_static("ratelimit-remaining"), http::HeaderName::from_static("ratelimit-reset"), http::HeaderName::from_static("ratelimit-policy"), http::header::RETRY_AFTER, ]), ) .with_state(state);
if cfg!(feature = "frontend") && tranquil_config::get().frontend.enabled { let frontend_dir = &tranquil_config::get().frontend.dir; let index_path = format!("{}/index.html", frontend_dir); let homepage_path = format!("{}/homepage.html", frontend_dir);
let homepage_exists = std::path::Path::new(&homepage_path).exists(); let homepage_file = if homepage_exists { homepage_path } else { index_path.clone() };
let spa_router = Router::new().fallback_service(ServeFile::new(&index_path));
let serve_dir = ServeDir::new(frontend_dir).not_found_service(ServeFile::new(&index_path));
return router .route_service("/", ServeFile::new(&homepage_file)) .nest("/app", spa_router) .fallback_service(serve_dir); }
router}
fn is_plain_text(headers: &http::HeaderMap) -> bool { headers .get(http::header::CONTENT_TYPE) .and_then(|v| v.to_str().ok()) .is_some_and(|ct| ct.starts_with("text/plain"))}
fn should_rewrite_to_xrpc_error(response: &axum::response::Response) -> bool { match response.status() { StatusCode::UNPROCESSABLE_ENTITY => true, StatusCode::BAD_REQUEST => is_plain_text(response.headers()), StatusCode::UNSUPPORTED_MEDIA_TYPE => is_plain_text(response.headers()), _ => false, }}
async fn rewrite_extractor_errors(response: axum::response::Response) -> axum::response::Response { if !should_rewrite_to_xrpc_error(&response) { return response; } let (mut parts, body) = response.into_parts(); let bytes = match axum::body::to_bytes(body, 64 * 1024).await { Ok(b) => b, Err(_) => { parts.status = StatusCode::BAD_REQUEST; parts.headers.remove(http::header::CONTENT_LENGTH); let fallback = json!({"error": "InvalidRequest", "message": "Invalid request body"}); return axum::response::Response::from_parts( parts, axum::body::Body::from(serde_json::to_vec(&fallback).unwrap_or_default()), ); } }; let raw = serde_json::from_slice::<serde_json::Value>(&bytes) .ok() .and_then(|v| v.get("message").and_then(|m| m.as_str()).map(String::from)) .unwrap_or_else(|| { String::from_utf8(bytes.to_vec()).unwrap_or_else(|_| "Invalid request body".into()) }); let message = humanize_extraction_error(&raw);
parts.status = StatusCode::BAD_REQUEST; parts.headers.remove(http::header::CONTENT_LENGTH); let error_name = classify_extraction_error(&raw); let new_body = json!({ "error": error_name, "message": message }); axum::response::Response::from_parts( parts, axum::body::Body::from(serde_json::to_vec(&new_body).unwrap_or_default()), )}
fn humanize_extraction_error(raw: &str) -> String { if raw.contains("missing field") { raw.split("missing field `") .nth(1) .and_then(|s| s.split('`').next()) .map(|field| format!("Missing required field: {}", field)) .unwrap_or_else(|| raw.to_string()) } else if raw.contains("invalid type") { format!("Invalid field type: {}", raw) } else if raw.contains("Invalid JSON") || raw.contains("syntax") { "Invalid JSON syntax".to_string() } else if raw.contains("Content-Type") || raw.contains("content type") { "Content-Type must be application/json".to_string() } else if raw.contains("Failed to parse") || raw.contains("expected ident") { "Invalid JSON in request body".to_string() } else if raw.contains("Failed to deserialize query string") { raw.strip_prefix("Failed to deserialize query string: ") .map(|rest| format!("Invalid query parameter: {}", rest)) .unwrap_or_else(|| "Invalid query parameters".into()) } else { raw.to_string() }}
fn classify_extraction_error(raw: &str) -> &'static str { match raw { s if s.contains("invalid handle") => "InvalidHandle", s if s.contains("invalid CID") || s.contains("invalid cid") => "InvalidRequest", _ => "InvalidRequest", }}