diff --git a/Cargo.lock b/Cargo.lock index 15ab551..157af95 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5478,6 +5478,8 @@ dependencies = [ "sha2 0.11.0", "shared", "tokio", + "tower", + "tower-http 0.6.11", "tracing", "tracing-subscriber", "unicode-normalization", diff --git a/Cargo.toml b/Cargo.toml index 5944907..b469487 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ sha2 = "0.11.0" shared = { path = "shared" } terminal_size = "0.4.4" tokio = { version = "1.53.1", features = ["macros", "rt-multi-thread"] } +tower-http = { version = "0.6", features = ["cors"] } tracing = "0.1.44" tracing-subscriber = "0.3.23" unicode-normalization = "0.1" diff --git a/server/Cargo.toml b/server/Cargo.toml index 7c69c53..119fdaa 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -11,7 +11,7 @@ base64.workspace = true chrono.workspace = true futures-util.workspace = true h3o.workspace = true -jacquard = { workspace = true, features = ["derive", "streaming", "zstd"] } +jacquard = { workspace = true, features = ["api_bluesky", "derive", "streaming", "zstd"] } jacquard-axum.workspace = true lexicons = { path = "../lexicons" } reqwest.workspace = true @@ -21,7 +21,11 @@ serde_json.workspace = true sha2.workspace = true shared.workspace = true tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] } +tower-http.workspace = true tracing.workspace = true tracing-subscriber.workspace = true unicode-normalization.workspace = true zip.workspace = true + +[dev-dependencies] +tower = { version = "0.5", features = ["util"] } diff --git a/server/src/handlers.rs b/server/src/handlers.rs index 93583cf..6a640c1 100644 --- a/server/src/handlers.rs +++ b/server/src/handlers.rs @@ -1,12 +1,29 @@ use crate::db::Database; use axum::{ - extract::{MatchedPath, Query, State}, + extract::State, http::{HeaderMap, HeaderValue, StatusCode}, response::IntoResponse, - routing::{get, post}, + routing::get, Router, }; -use serde::Deserialize; +use jacquard::api::{ + app_bsky::actor::search_actors_typeahead::SearchActorsTypeaheadRequest, + com_atproto::identity::resolve_handle::ResolveHandleRequest, + com_atproto::repo::{ + create_record::{CreateRecord, CreateRecordRequest}, + delete_record::{DeleteRecord, DeleteRecordRequest}, + get_record::GetRecordRequest, + list_records::ListRecordsRequest, + put_record::{PutRecord, PutRecordRequest}, + }, +}; +use jacquard::xrpc::XrpcRequest; +use jacquard_axum::{ExtractXrpc, IntoRouter}; +use lexicons::fyi_wed::{ + get_events_skeleton::GetEventsSkeletonRequest, get_forecast::GetForecastRequest, + resolve_geolocation::ResolveGeolocationRequest, + search_locations_typeahead::SearchLocationsTypeaheadRequest, +}; use serde_json::Value; use sha2::{Digest, Sha256}; use shared::{ @@ -24,6 +41,7 @@ use std::{ }, time::Duration, }; +use tower_http::cors::{AllowOrigin, CorsLayer}; #[derive(Clone)] pub struct RateLimiter { @@ -59,6 +77,20 @@ pub struct AppState { pub jetstream_connected: Arc, } +pub fn cors_layer(allowed_origins: &HashSet) -> CorsLayer { + CorsLayer::new() + .allow_origin(AllowOrigin::list( + allowed_origins.iter().filter_map(|o| o.parse().ok()), + )) + .allow_methods([axum::http::Method::GET, axum::http::Method::POST]) + .allow_headers([ + axum::http::header::AUTHORIZATION, + axum::http::header::CONTENT_TYPE, + axum::http::HeaderName::from_static("dpop"), + ]) + .expose_headers([axum::http::HeaderName::from_static("dpop-nonce")]) +} + fn cache_key(event: &Event, units: &Units) -> String { let mut hasher = Sha256::new(); hasher.update(event.when.to_string().as_bytes()); @@ -91,64 +123,60 @@ fn chart_format_from(format: Option<&str>) -> Option { } } -#[derive(Deserialize)] -struct ResolveHandleQuery { - handle: String, -} - async fn handle( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { let client = state.client.raw(); - match shared::atproto::client::resolve_handle(client, &query.handle).await { - Ok(did) => with_cors_json( - serde_json::json!({"did":did}), - &headers, - &state.allowed_origins, - ), - Err(_) => with_cors( + let handle_str = req.handle.as_ref(); + match shared::atproto::client::resolve_handle(client, handle_str).await { + Ok(did) => axum::Json(serde_json::json!({"did":did})).into_response(), + Err(_) => ( StatusCode::NOT_FOUND, axum::Json(serde_json::json!({"error":"could not resolve handle"})), - &headers, - &state.allowed_origins, - ), + ) + .into_response(), } } const BSKY_PUBLIC_API: &str = "https://public.api.bsky.app"; const TYPEAHEAD_LIMIT: usize = 7; -#[derive(Deserialize)] -struct TypeaheadQuery { - q: String, -} - async fn actors_typeahead( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { let unavailable = |status: StatusCode, e: &dyn std::fmt::Display| { tracing::warn!("actors typeahead: {e}"); - with_cors( + ( status, axum::Json(serde_json::json!({"error":"typeahead unavailable"})), - &headers, - &state.allowed_origins, ) }; + let q = match req.q.as_ref() { + Some(q) if !q.is_empty() => q.as_ref() as &str, + _ => { + return ( + StatusCode::BAD_REQUEST, + axum::Json(serde_json::json!({"error":"missing query"})), + ) + .into_response() + } + }; let mut url = reqwest::Url::parse(BSKY_PUBLIC_API).expect("valid base url"); url.set_path("/xrpc/app.bsky.actor.searchActorsTypeahead"); url.query_pairs_mut() - .append_pair("q", &query.q) + .append_pair("q", q) .append_pair("limit", &TYPEAHEAD_LIMIT.to_string()); let resp = match tokio::time::timeout(Duration::from_secs(5), state.client.get(url)).await { Ok(Ok(resp)) if resp.status().is_success() => resp, - Ok(Ok(resp)) => return unavailable(StatusCode::BAD_GATEWAY, &resp.status()), - Ok(Err(e)) => return unavailable(StatusCode::BAD_GATEWAY, &e), - Err(_) => return unavailable(StatusCode::GATEWAY_TIMEOUT, &"upstream timed out"), + Ok(Ok(resp)) => { + return unavailable(StatusCode::BAD_GATEWAY, &resp.status()).into_response() + } + Ok(Err(e)) => return unavailable(StatusCode::BAD_GATEWAY, &e).into_response(), + Err(_) => { + return unavailable(StatusCode::GATEWAY_TIMEOUT, &"upstream timed out").into_response() + } }; let is_json = resp .headers() @@ -157,66 +185,58 @@ async fn actors_typeahead( .map(|v| v.contains("json")) .unwrap_or(false); if !is_json { - return unavailable(StatusCode::BAD_GATEWAY, &"upstream returned non-JSON"); + return unavailable(StatusCode::BAD_GATEWAY, &"upstream returned non-JSON").into_response(); } match resp.json::().await { - Ok(body) => with_cors_json(body, &headers, &state.allowed_origins), - Err(e) => unavailable(StatusCode::BAD_GATEWAY, &e), + Ok(body) => axum::Json(body).into_response(), + Err(e) => unavailable(StatusCode::BAD_GATEWAY, &e).into_response(), } } -#[derive(Deserialize)] -struct GetRecordQuery { - repo: String, - collection: String, - rkey: String, -} - async fn record( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { let client = state.client.raw(); - let pds = match shared::atproto::client::resolve_pds(client, &query.repo).await { + let repo = req.repo.as_ref().to_string(); + let collection = req.collection.as_ref().to_string(); + let rkey = req.rkey.as_ref().to_string(); + let pds = match shared::atproto::client::resolve_pds(client, &repo).await { Ok(pds) => pds, Err(_) => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"could not resolve PDS"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; let base = match shared::atproto::client::pds_xrpc_url(&pds, "com.atproto.repo.getRecord") { Ok(url) => url, Err(_) => { - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"invalid PDS URL"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; let url = match reqwest::Url::parse_with_params( base.as_str(), &[ - ("repo", query.repo.as_str()), - ("collection", query.collection.as_str()), - ("rkey", query.rkey.as_str()), + ("repo", repo.as_str()), + ("collection", collection.as_str()), + ("rkey", rkey.as_str()), ], ) { Ok(url) => url, Err(e) => { tracing::error!("invalid getRecord params: {e}"); - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"invalid upstream request"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } }; match client.get(url).send().await { @@ -224,92 +244,68 @@ async fn record( if !resp.status().is_success() { let status = resp.status(); tracing::error!("getRecord upstream returned {status}"); - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"upstream error"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } if let Ok(json) = resp.json::().await { - with_cors_json(json, &headers, &state.allowed_origins) + axum::Json(json).into_response() } else { - with_cors( - StatusCode::INTERNAL_SERVER_ERROR, - "json parse", - &headers, - &state.allowed_origins, - ) + (StatusCode::INTERNAL_SERVER_ERROR, "json parse").into_response() } } - Err(_) => with_cors( - StatusCode::BAD_GATEWAY, - "upstream", - &headers, - &state.allowed_origins, - ), + Err(_) => (StatusCode::BAD_GATEWAY, "upstream").into_response(), } } -#[derive(Deserialize)] -struct ListRecordsQuery { - repo: String, - collection: String, - limit: Option, - cursor: Option, -} - async fn records( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { let client = state.client.raw(); - let pds = match shared::atproto::client::resolve_pds(client, &query.repo).await { + let repo = req.repo.as_ref().to_string(); + let collection = req.collection.as_ref().to_string(); + let pds = match shared::atproto::client::resolve_pds(client, &repo).await { Ok(pds) => pds, Err(_) => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"could not resolve PDS"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; - let mut params: Vec<(&str, &str)> = vec![ - ("repo", query.repo.as_str()), - ("collection", query.collection.as_str()), - ]; + let mut params: Vec<(&str, &str)> = + vec![("repo", repo.as_str()), ("collection", collection.as_str())]; let limit_str; - if let Some(limit) = &query.limit { + if let Some(limit) = &req.limit { limit_str = limit.to_string(); params.push(("limit", limit_str.as_str())); } - let cursor = query.cursor; - if let Some(cursor_value) = &cursor { - params.push(("cursor", cursor_value.as_str())); + if let Some(cursor) = &req.cursor { + params.push(("cursor", cursor.as_ref())); } let base = match shared::atproto::client::pds_xrpc_url(&pds, "com.atproto.repo.listRecords") { Ok(url) => url, Err(_) => { - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"invalid PDS URL"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; let url = match reqwest::Url::parse_with_params(base.as_str(), ¶ms) { Ok(url) => url, Err(e) => { tracing::error!("invalid listRecords params: {e}"); - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"invalid upstream request"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } }; match client.get(url).send().await { @@ -317,97 +313,126 @@ async fn records( if !resp.status().is_success() { let status = resp.status(); tracing::error!("listRecords upstream returned {status}"); - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"upstream error"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } if let Ok(json) = resp.json::().await { - with_cors_json(json, &headers, &state.allowed_origins) + axum::Json(json).into_response() } else { - with_cors( - StatusCode::INTERNAL_SERVER_ERROR, - "json parse", - &headers, - &state.allowed_origins, - ) + (StatusCode::INTERNAL_SERVER_ERROR, "json parse").into_response() } } - Err(_) => with_cors( - StatusCode::BAD_GATEWAY, - "upstream", - &headers, - &state.allowed_origins, - ), + Err(_) => (StatusCode::BAD_GATEWAY, "upstream").into_response(), } } -async fn write_record( +async fn create_record( + State(state): State, + headers: HeaderMap, + axum::Json(body): axum::Json, +) -> axum::response::Response { + forward_write( + &state, + &headers, + CreateRecord::::NSID, + &body, + true, + ) + .await +} + +async fn put_record( State(state): State, - matched: MatchedPath, headers: HeaderMap, axum::Json(body): axum::Json, ) -> axum::response::Response { - let method = matched.as_str().trim_start_matches("/xrpc/"); + forward_write( + &state, + &headers, + PutRecord::::NSID, + &body, + true, + ) + .await +} + +async fn delete_record( + State(state): State, + headers: HeaderMap, + axum::Json(body): axum::Json, +) -> axum::response::Response { + forward_write( + &state, + &headers, + DeleteRecord::::NSID, + &body, + false, + ) + .await +} + +async fn forward_write( + state: &AppState, + headers: &HeaderMap, + method: &str, + body: &Value, + validate: bool, +) -> axum::response::Response { let repo = match body.get("repo").and_then(Value::as_str) { Some(repo) => repo, None => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"missing repo"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; let client = state.client.raw(); let pds = match shared::atproto::client::resolve_pds(client, repo).await { Ok(pds) => pds, Err(_) => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"could not resolve PDS"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; let url = match shared::atproto::client::pds_xrpc_url(&pds, method) { Ok(url) => url, Err(_) => { - return with_cors( + return ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"invalid PDS URL"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; - let mut req = client.post(url).json(&body); - if method == "com.atproto.repo.createRecord" || method == "com.atproto.repo.putRecord" { + let mut req = client.post(url).json(body); + if validate { let collection = body.get("collection").and_then(Value::as_str); if collection == Some(EVENT) { if let Some(record) = body.get("record") { match serde_json::from_value::(record.clone()) { Ok(event) => { if let Err(e) = event.validate() { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error": e.to_string()})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } } Err(_) => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"invalid record"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } } } @@ -416,21 +441,19 @@ async fn write_record( match serde_json::from_value::(record.clone()) { Ok(follow) => { if let Err(e) = state.client.validate_follow(&follow).await { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error": e.to_string()})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } } Err(_) => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"invalid follow record"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } } } @@ -464,8 +487,7 @@ async fn write_record( } let status = StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - let mut response = - with_cors(status, axum::Json(json), &headers, &state.allowed_origins); + let mut response = (status, axum::Json(json)).into_response(); if let Some(nonce) = nonce { if let Ok(value) = HeaderValue::from_str(&nonce) { response @@ -477,112 +499,85 @@ async fn write_record( } Err(e) => { tracing::error!("{method} upstream request failed: {e}"); - with_cors( + ( StatusCode::BAD_GATEWAY, axum::Json(serde_json::json!({"error":"upstream"})), - &headers, - &state.allowed_origins, ) + .into_response() } } } -#[derive(Deserialize)] -struct GetForecastQuery { - uri: String, - units: Option, - format: Option, -} - async fn forecast( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { - let uri = if !query.uri.starts_with("at://") { - format!("at://{}", query.uri) + let uri_str: &str = req.uri.as_ref(); + let uri = if !uri_str.starts_with("at://") { + format!("at://{}", uri_str) } else { - query.uri + uri_str.to_string() }; let (_, collection, _) = match parse_at_uri(&uri) { Ok(parsed) => parsed, Err(_) => { - return with_cors( + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"invalid uri"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; let event = match collection { EVENT => match shared::atproto::client::event(&uri).await { Ok(event) => event, Err(_) => { - return with_cors( + return ( StatusCode::NOT_FOUND, axum::Json(serde_json::json!({"error":"not found"})), - &headers, - &state.allowed_origins, ) + .into_response() } }, CALENDAR => { let calendar_event = match shared::atproto::client::calendar_event(&uri).await { Ok(event) => event, Err(_) => { - return with_cors( + return ( StatusCode::NOT_FOUND, axum::Json(serde_json::json!({"error":"not found"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; match shared::atproto::client::calendar_into_event(calendar_event, &state.client).await { Ok(event) => event, Err(_) => { - return with_cors( + return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({"error":"invalid event"})), - &headers, - &state.allowed_origins, ) + .into_response() } } } _ => { - return with_cors( + return ( StatusCode::NOT_FOUND, axum::Json(serde_json::json!({"error":"not found"})), - &headers, - &state.allowed_origins, ) + .into_response() } }; - let units = units_from(query.units.as_deref()); - if let Some(format) = chart_format_from(query.format.as_deref()) { - return chart( - &state, - &event, - &units, - &format, - &headers, - &state.allowed_origins, - ) - .await; + let units = units_from(req.units.as_ref().map(|u| u.as_str())); + if let Some(format) = chart_format_from(req.format.as_ref().map(|f| f.as_str())) { + return chart(&state, &event, &units, &format).await; } - weather(&state, &event, &units, &headers, &state.allowed_origins).await + weather(&state, &event, &units).await } -async fn weather( - state: &AppState, - event: &Event, - units: &Units, - headers: &HeaderMap, - allowed_origins: &HashSet, -) -> axum::response::Response { +async fn weather(state: &AppState, event: &Event, units: &Units) -> axum::response::Response { let hash = cache_key(event, units); if !shared::weather::has_forecast(event, false) { let message = if (event.when - chrono::Local::now().naive_local()).num_days() < 0 { @@ -590,12 +585,11 @@ async fn weather( } else { "too far" }; - return with_cors( + return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({"error":message})), - headers, - allowed_origins, - ); + ) + .into_response(); } match state.db.weather( &hash, @@ -605,7 +599,7 @@ async fn weather( state.ttl.as_secs(), ) { Ok(data) => match serde_json::from_slice::(&data) { - Ok(json) => return with_cors_json(json, headers, allowed_origins), + Ok(json) => return axum::Json(json).into_response(), Err(e) => tracing::warn!("cache read: corrupt JSON for {hash}: {e}"), }, Err(e) => tracing::warn!("cache read failed for {hash}: {e}"), @@ -614,14 +608,7 @@ async fn weather( let weather = match shared::weather::weather_from(event, &state.client, units, Some(window)).await { Ok(weather) => weather, - Err(_) => { - return with_cors( - StatusCode::FAILED_DEPENDENCY, - "upstream", - headers, - allowed_origins, - ) - } + Err(_) => return (StatusCode::FAILED_DEPENDENCY, "upstream").into_response(), }; let value = serde_json::to_value(&weather).unwrap_or(Value::Null); if let Ok(bytes) = serde_json::to_vec(&value) { @@ -636,7 +623,7 @@ async fn weather( ); } } - with_cors_json(value, headers, allowed_origins) + axum::Json(value).into_response() } async fn chart( @@ -644,40 +631,21 @@ async fn chart( event: &Event, units: &Units, format: &Format, - headers: &HeaderMap, - allowed_origins: &HashSet, ) -> axum::response::Response { if !shared::weather::has_forecast(event, false) { - return with_cors( - StatusCode::UNPROCESSABLE_ENTITY, - "no forecast", - headers, - allowed_origins, - ); + return (StatusCode::UNPROCESSABLE_ENTITY, "no forecast").into_response(); } let window = shared::weather::window_from(event); let weather = match shared::weather::weather_from(event, &state.client, units, Some(window)).await { Ok(weather) => weather, - Err(_) => { - return with_cors( - StatusCode::FAILED_DEPENDENCY, - "upstream", - headers, - allowed_origins, - ) - } + Err(_) => return (StatusCode::FAILED_DEPENDENCY, "upstream").into_response(), }; let body = match shared::chart::render(&weather, *format) { Ok((body, _)) => body, Err(e) => { tracing::warn!("chart: {e}"); - return with_cors( - StatusCode::UNPROCESSABLE_ENTITY, - "no forecast", - headers, - allowed_origins, - ); + return (StatusCode::UNPROCESSABLE_ENTITY, "no forecast").into_response(); } }; let mut response = (StatusCode::OK, body).into_response(); @@ -685,16 +653,9 @@ async fn chart( axum::http::header::CONTENT_TYPE, HeaderValue::from_static(format.content_type()), ); - add_cors_headers(&mut response, headers, allowed_origins); response } -#[derive(Deserialize)] -struct ResolveGeoQuery { - #[serde(rename = "query")] - value: String, -} - fn geo_json(latitude: f64, longitude: f64) -> Value { serde_json::json!({ "latitude": (latitude * GEOCOORDINATE_SCALE).round() as i64, @@ -714,172 +675,89 @@ fn location_to_json(loc: &crate::db::Location) -> Value { async fn search_locations_typeahead( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { - if query.value.chars().count() > shared::atproto::LOCATION_MAX_LEN { - return with_cors( + let query: &str = req.query.as_ref(); + if query.chars().count() > shared::atproto::LOCATION_MAX_LEN { + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"query too long"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } - match state.db.search_locations(&query.value, TYPEAHEAD_LIMIT) { + match state.db.search_locations(query, TYPEAHEAD_LIMIT) { Ok(locations) => { let locations: Vec = locations.iter().map(location_to_json).collect(); - with_cors_json( - serde_json::json!({ "locations": locations }), - &headers, - &state.allowed_origins, - ) + axum::Json(serde_json::json!({ "locations": locations })).into_response() } Err(e) => { tracing::error!("search locations: {e}"); - with_cors( + ( StatusCode::INTERNAL_SERVER_ERROR, axum::Json(serde_json::json!({"error":"search failed"})), - &headers, - &state.allowed_origins, ) + .into_response() } } } async fn geo( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { - if query.value.chars().count() > shared::atproto::LOCATION_MAX_LEN { - return with_cors( + let query: &str = req.query.as_ref(); + if query.chars().count() > shared::atproto::LOCATION_MAX_LEN { + return ( StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({"error":"query too long"})), - &headers, - &state.allowed_origins, - ); + ) + .into_response(); } - match state.db.geo(&query.value) { - Ok((lat, lon)) => { - return with_cors_json(geo_json(lat, lon), &headers, &state.allowed_origins) - } + match state.db.geo(query) { + Ok((lat, lon)) => return axum::Json(geo_json(lat, lon)).into_response(), Err(e) => tracing::warn!("geo cache read: {e}"), } + let query_owned = query.to_string(); state.rate_limiter.wait().await; - match shared::geo::fetch::coordinates_from_nominatim(&state.client, &query.value).await { + match shared::geo::fetch::coordinates_from_nominatim(&state.client, &query_owned).await { Ok((lat, lon)) => { - if let Err(e) = state.db.save_geo(&query.value, lat, lon) { + if let Err(e) = state.db.save_geo(&query_owned, lat, lon) { tracing::error!("geo: {e}"); } - with_cors_json(geo_json(lat, lon), &headers, &state.allowed_origins) + axum::Json(geo_json(lat, lon)).into_response() } Err(e) => { - tracing::warn!("geo upstream error for {:?}: {e:#}", query.value); - with_cors( + tracing::warn!("geo upstream error for {:?}: {e:#}", query_owned); + ( StatusCode::NOT_FOUND, axum::Json(serde_json::json!({"error":"not found"})), - &headers, - &state.allowed_origins, ) + .into_response() } } } -#[derive(Deserialize)] -struct ListFollowedEventsQuery { - did: String, -} - async fn followed_events( State(state): State, - headers: HeaderMap, - Query(query): Query, + ExtractXrpc(req): ExtractXrpc, ) -> axum::response::Response { - match state.db.follow_uris(&query.did) { + let did = req.did.as_ref(); + match state.db.follow_uris(did) { Ok(uris) => { let uris: Vec<&str> = uris.iter().map(|(_, uri)| uri.as_str()).collect(); - with_cors_json( - serde_json::json!({ "uris": uris }), - &headers, - &state.allowed_origins, - ) + axum::Json(serde_json::json!({ "uris": uris })).into_response() } Err(e) => { tracing::error!("getEventsSkeleton: {e}"); - with_cors( + ( StatusCode::INTERNAL_SERVER_ERROR, axum::Json(serde_json::json!({"error":"database error"})), - &headers, - &state.allowed_origins, ) + .into_response() } } } -fn with_cors( - status: StatusCode, - body: T, - headers: &HeaderMap, - allowed_origins: &HashSet, -) -> axum::response::Response { - let mut response = (status, body).into_response(); - add_cors_headers(&mut response, headers, allowed_origins); - response -} - -fn with_cors_json( - value: Value, - headers: &HeaderMap, - allowed_origins: &HashSet, -) -> axum::response::Response { - let mut response = axum::Json(value).into_response(); - add_cors_headers(&mut response, headers, allowed_origins); - response -} - -fn add_cors_headers( - response: &mut axum::response::Response, - headers: &HeaderMap, - allowed_origins: &HashSet, -) { - let origin = match headers - .get(axum::http::header::ORIGIN) - .and_then(|origin| origin.to_str().ok()) - { - Some(origin) if allowed_origins.contains(origin) => origin, - _ => return, - }; - if let Ok(origin_value) = HeaderValue::from_str(origin) { - response.headers_mut().insert( - axum::http::header::ACCESS_CONTROL_ALLOW_ORIGIN, - origin_value, - ); - response.headers_mut().insert( - axum::http::header::ACCESS_CONTROL_ALLOW_METHODS, - HeaderValue::from_static("GET, POST, OPTIONS"), - ); - response.headers_mut().insert( - axum::http::header::ACCESS_CONTROL_ALLOW_HEADERS, - HeaderValue::from_static("Authorization, Content-Type, DPoP"), - ); - response.headers_mut().insert( - axum::http::header::ACCESS_CONTROL_EXPOSE_HEADERS, - HeaderValue::from_static("DPoP-Nonce"), - ); - } else { - tracing::warn!("skipping CORS headers: invalid origin value {origin}"); - } -} - -async fn options_handler( - State(state): State, - headers: HeaderMap, -) -> axum::response::Response { - let mut response = (StatusCode::OK).into_response(); - add_cors_headers(&mut response, &headers, &state.allowed_origins); - response -} - async fn health() -> StatusCode { StatusCode::OK } @@ -893,61 +771,34 @@ async fn ready(State(state): State) -> StatusCode { } pub fn xrpc_router(state: AppState) -> Router { + let cors = cors_layer(&state.allowed_origins); Router::new() .route("/health", get(health)) .route("/ready", get(ready)) - .route( - "/xrpc/com.atproto.identity.resolveHandle", - get(handle).options(options_handler), - ) - .route( - "/xrpc/app.bsky.actor.searchActorsTypeahead", - get(actors_typeahead).options(options_handler), - ) - .route( - "/xrpc/com.atproto.repo.getRecord", - get(record).options(options_handler), - ) - .route( - "/xrpc/com.atproto.repo.listRecords", - get(records).options(options_handler), - ) - .route( - "/xrpc/com.atproto.repo.createRecord", - post(write_record).options(options_handler), - ) - .route( - "/xrpc/com.atproto.repo.putRecord", - post(write_record).options(options_handler), - ) - .route( - "/xrpc/com.atproto.repo.deleteRecord", - post(write_record).options(options_handler), - ) - .route( - "/xrpc/fyi.wed.getForecast", - get(forecast).options(options_handler), - ) - .route( - "/xrpc/fyi.wed.resolveGeolocation", - get(geo).options(options_handler), - ) - .route( - "/xrpc/fyi.wed.searchLocationsTypeahead", - get(search_locations_typeahead).options(options_handler), - ) - .route( - "/xrpc/fyi.wed.getEventsSkeleton", - get(followed_events).options(options_handler), - ) + .merge(ResolveHandleRequest::into_router(handle)) + .merge(SearchActorsTypeaheadRequest::into_router(actors_typeahead)) + .merge(GetRecordRequest::into_router(record)) + .merge(ListRecordsRequest::into_router(records)) + .merge(CreateRecordRequest::into_router(create_record)) + .merge(PutRecordRequest::into_router(put_record)) + .merge(DeleteRecordRequest::into_router(delete_record)) + .merge(GetForecastRequest::into_router(forecast)) + .merge(ResolveGeolocationRequest::into_router(geo)) + .merge(SearchLocationsTypeaheadRequest::into_router( + search_locations_typeahead, + )) + .merge(GetEventsSkeletonRequest::into_router(followed_events)) + .layer(cors) .with_state(state) } #[cfg(test)] mod tests { use super::*; - use axum::http::{header::ORIGIN, HeaderValue}; + use axum::http::header::ORIGIN; use chrono::NaiveDateTime; + use lexicons::fyi_wed::resolve_geolocation::ResolveGeolocation; + use lexicons::fyi_wed::search_locations_typeahead::SearchLocationsTypeahead; use std::time::{SystemTime, UNIX_EPOCH}; const ALLOWED_ORIGIN: &str = "https://wed.fyi"; @@ -965,6 +816,10 @@ mod tests { } fn test_state(path: std::path::PathBuf) -> AppState { + test_state_with(path, HashSet::from([ALLOWED_ORIGIN.to_string()])) + } + + fn test_state_with(path: std::path::PathBuf, allowed_origins: HashSet) -> AppState { let db = Database::open(&path, 1).unwrap(); AppState { db: Arc::new(db), @@ -972,7 +827,7 @@ mod tests { ttl: Duration::from_secs(3600), radius: 15.0, rate_limiter: RateLimiter::new(Duration::from_millis(0)), - allowed_origins: HashSet::from([ALLOWED_ORIGIN.to_string()]), + allowed_origins, jetstream_connected: Arc::new(AtomicBool::new(true)), } } @@ -1087,23 +942,21 @@ mod tests { assert_eq!(Format::Png.content_type(), "image/png"); } - fn response_with_origin(origin: Option<&str>) -> axum::response::Response { - let mut response = StatusCode::OK.into_response(); - let mut headers = HeaderMap::new(); - if let Some(origin) = origin { - headers.insert(ORIGIN, HeaderValue::from_str(origin).unwrap()); - } - add_cors_headers( - &mut response, - &headers, - &HashSet::from([ALLOWED_ORIGIN.to_string()]), - ); - response - } - - #[test] - fn cors_headers_set_for_allowed_origin() { - let response = response_with_origin(Some(ALLOWED_ORIGIN)); + #[tokio::test] + async fn cors_allows_listed_origin() { + let path = temp_db_path(); + let cleanup = DbCleanup(path.clone()); + let state = test_state(path); + state.db.save_geo("Berlin", 52.52, 13.405).unwrap(); + let app = xrpc_router(state); + let request = axum::http::Request::builder() + .method("GET") + .uri("/xrpc/fyi.wed.resolveGeolocation?query=Berlin") + .header(ORIGIN, ALLOWED_ORIGIN) + .body(axum::body::Body::empty()) + .unwrap(); + let response = tower::ServiceExt::oneshot(app, request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); assert_eq!( response .headers() @@ -1111,42 +964,98 @@ mod tests { .and_then(|v| v.to_str().ok()), Some(ALLOWED_ORIGIN) ); - assert!(response - .headers() - .contains_key(axum::http::header::ACCESS_CONTROL_ALLOW_METHODS)); - assert!(response - .headers() - .contains_key(axum::http::header::ACCESS_CONTROL_ALLOW_HEADERS)); + drop(cleanup); } - #[test] - fn cors_headers_absent_for_disallowed_origin() { - let response = response_with_origin(Some("https://evil.example")); + #[tokio::test] + async fn cors_blocks_disallowed_origin() { + let path = temp_db_path(); + let cleanup = DbCleanup(path.clone()); + let state = test_state(path); + state.db.save_geo("Berlin", 52.52, 13.405).unwrap(); + let app = xrpc_router(state); + let request = axum::http::Request::builder() + .method("GET") + .uri("/xrpc/fyi.wed.resolveGeolocation?query=Berlin") + .header(ORIGIN, "https://evil.example") + .body(axum::body::Body::empty()) + .unwrap(); + let response = tower::ServiceExt::oneshot(app, request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); assert!(response .headers() .get(axum::http::header::ACCESS_CONTROL_ALLOW_ORIGIN) .is_none()); + drop(cleanup); } - #[test] - fn cors_headers_absent_without_origin() { - let response = response_with_origin(None); + #[tokio::test] + async fn cors_blocks_missing_origin() { + let path = temp_db_path(); + let cleanup = DbCleanup(path.clone()); + let state = test_state(path); + state.db.save_geo("Berlin", 52.52, 13.405).unwrap(); + let app = xrpc_router(state); + let request = axum::http::Request::builder() + .method("GET") + .uri("/xrpc/fyi.wed.resolveGeolocation?query=Berlin") + .body(axum::body::Body::empty()) + .unwrap(); + let response = tower::ServiceExt::oneshot(app, request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); assert!(response .headers() .get(axum::http::header::ACCESS_CONTROL_ALLOW_ORIGIN) .is_none()); + drop(cleanup); } - #[test] - fn cors_headers_absent_for_empty_allowlist() { - let mut response = StatusCode::OK.into_response(); - let mut headers = HeaderMap::new(); - headers.insert(ORIGIN, HeaderValue::from_str(ALLOWED_ORIGIN).unwrap()); - add_cors_headers(&mut response, &headers, &HashSet::new()); + #[tokio::test] + async fn cors_preflight_returns_allow_methods_and_headers() { + let path = temp_db_path(); + let cleanup = DbCleanup(path.clone()); + let state = test_state(path); + let app = xrpc_router(state); + let request = axum::http::Request::builder() + .method("OPTIONS") + .uri("/xrpc/fyi.wed.resolveGeolocation?query=Berlin") + .header(ORIGIN, ALLOWED_ORIGIN) + .header(axum::http::header::ACCESS_CONTROL_REQUEST_METHOD, "GET") + .body(axum::body::Body::empty()) + .unwrap(); + let response = tower::ServiceExt::oneshot(app, request).await.unwrap(); + assert!(response + .headers() + .contains_key(axum::http::header::ACCESS_CONTROL_ALLOW_METHODS)); + assert!(response + .headers() + .contains_key(axum::http::header::ACCESS_CONTROL_ALLOW_HEADERS)); + assert!(response + .headers() + .contains_key(axum::http::header::ACCESS_CONTROL_ALLOW_ORIGIN)); + drop(cleanup); + } + + #[tokio::test] + async fn cors_empty_allowlist_blocks_all() { + let path = temp_db_path(); + let cleanup = DbCleanup(path.clone()); + let state = test_state_with(path, HashSet::new()); + state.db.save_geo("Berlin", 52.52, 13.405).unwrap(); + let app = xrpc_router(state); + let request = axum::http::Request::builder() + .method("GET") + .uri("/xrpc/fyi.wed.resolveGeolocation?query=Berlin") + .header(ORIGIN, ALLOWED_ORIGIN) + .body(axum::body::Body::empty()) + .unwrap(); + let response = tower::ServiceExt::oneshot(app, request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); assert!(response .headers() .get(axum::http::header::ACCESS_CONTROL_ALLOW_ORIGIN) .is_none()); + drop(cleanup); } #[tokio::test] @@ -1157,9 +1066,8 @@ mod tests { state.db.save_geo("Berlin", 52.52, 13.405).unwrap(); let response = geo( State(state), - HeaderMap::new(), - Query(ResolveGeoQuery { - value: "Berlin".to_string(), + ExtractXrpc(ResolveGeolocation { + query: "Berlin".into(), }), ) .await; @@ -1173,32 +1081,6 @@ mod tests { drop(cleanup); } - #[tokio::test] - async fn geo_includes_cors_for_allowed_origin() { - let path = temp_db_path(); - let cleanup = DbCleanup(path.clone()); - let state = test_state(path); - state.db.save_geo("Berlin", 52.52, 13.405).unwrap(); - let mut headers = HeaderMap::new(); - headers.insert(ORIGIN, HeaderValue::from_str(ALLOWED_ORIGIN).unwrap()); - let response = geo( - State(state), - headers, - Query(ResolveGeoQuery { - value: "Berlin".to_string(), - }), - ) - .await; - assert_eq!( - response - .headers() - .get(axum::http::header::ACCESS_CONTROL_ALLOW_ORIGIN) - .and_then(|v| v.to_str().ok()), - Some(ALLOWED_ORIGIN) - ); - drop(cleanup); - } - #[test] fn geo_json_scales_coordinates() { let json = geo_json(52.52, 13.405); @@ -1249,9 +1131,8 @@ mod tests { seed_locations(&state); let response = search_locations_typeahead( State(state), - HeaderMap::new(), - Query(ResolveGeoQuery { - value: "par".to_string(), + ExtractXrpc(SearchLocationsTypeahead { + query: "par".into(), }), ) .await; @@ -1276,9 +1157,8 @@ mod tests { seed_locations(&state); let response = search_locations_typeahead( State(state), - HeaderMap::new(), - Query(ResolveGeoQuery { - value: "paul".to_string(), + ExtractXrpc(SearchLocationsTypeahead { + query: "paul".into(), }), ) .await; @@ -1300,9 +1180,8 @@ mod tests { let state = test_state(path); let response = search_locations_typeahead( State(state), - HeaderMap::new(), - Query(ResolveGeoQuery { - value: "x".repeat(shared::atproto::LOCATION_MAX_LEN + 1), + ExtractXrpc(SearchLocationsTypeahead { + query: "x".repeat(shared::atproto::LOCATION_MAX_LEN + 1).into(), }), ) .await; diff --git a/spa/src/app/views.rs b/spa/src/app/views.rs index d9e53a9..dfa0e62 100644 --- a/spa/src/app/views.rs +++ b/spa/src/app/views.rs @@ -88,7 +88,6 @@ impl App { ().ok());