diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 2aebecfe5..394b3a010 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -4229,6 +4229,20 @@ export interface operations { }; content?: never; }; + /** @description Body over its wire or decompressed bound */ + 413: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Content encoding other than gzip */ + 415: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; }; }; metadata: { diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs index a4a82ccea..5b8f666c4 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs @@ -12,6 +12,7 @@ pub(crate) mod person_maintenance; pub(crate) mod person_node_sql; pub(crate) mod person_operations; pub(crate) mod person_reads; +pub(crate) mod request_json; pub(crate) mod router; pub(crate) mod run_reads; pub(crate) mod runs; diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/request_json.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/request_json.rs new file mode 100644 index 000000000..3956a45d3 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/request_json.rs @@ -0,0 +1,125 @@ +//! JSON request bodies that may arrive gzip-encoded, decoded within a decompressed-size bound. + +use std::io::{BufReader, Read}; + +use axum::http::{HeaderMap, StatusCode, header}; +use flate2::read::MultiGzDecoder; +use serde::de::DeserializeOwned; + +use crate::api::constants::{GZIP_ENCODING, IDENTITY_ENCODING, JSON_MEDIA_TYPE}; +use crate::api::errors::ApiError; +use crate::api::models::BoundedReader; + +/// `body` as JSON after its declared encoding; gzip streams into the parser and stops past `limit` bytes. +pub(crate) fn request_json( + headers: &HeaderMap, + body: &[u8], + limit: usize, +) -> Result { + if !json_content(headers) { + return Err(rejection( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "expected an application/json body".into(), + )); + } + match encodings(headers)?.as_slice() { + [] => serde_json::from_slice(body).map_err(|error| parse_rejection(&error)), + [encoding] if encoding == GZIP_ENCODING => { + let mut reader = BoundedReader { + inner: MultiGzDecoder::new(body), + remaining: limit, + exceeded: false, + }; + let parsed: Result = + serde_json::from_reader(BufReader::new(&mut reader)); + match parsed { + Ok(value) => Ok(value), + Err(_) if reader.exceeded => Err(rejection( + StatusCode::PAYLOAD_TOO_LARGE, + format!("decompressed body exceeds {limit} bytes"), + )), + Err(error) if error.is_io() => Err(rejection( + StatusCode::BAD_REQUEST, + "request body is not valid gzip".into(), + )), + Err(error) => Err(parse_rejection(&error)), + } + } + _ => Err(rejection( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "only gzip content encoding is accepted".into(), + )), + } +} + +impl Read for BoundedReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let window = buffer.len().min(self.remaining.saturating_add(1)); + let read = self.inner.read(&mut buffer[..window])?; + if read > self.remaining { + self.exceeded = true; + return Err(std::io::Error::other("decompressed body exceeds its bound")); + } + self.remaining -= read; + Ok(read) + } +} + +fn json_content(headers: &HeaderMap) -> bool { + let Some(value) = headers.get(header::CONTENT_TYPE) else { + return false; + }; + let Ok(text) = value.to_str() else { + return false; + }; + let essence: String = text + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + essence == JSON_MEDIA_TYPE + || (essence.starts_with("application/") && essence.ends_with("+json")) +} + +/// Declared encodings other than identity, in order; an unreadable header is refused. +fn encodings(headers: &HeaderMap) -> Result, ApiError> { + let mut encodings: Vec = Vec::new(); + for value in headers.get_all(header::CONTENT_ENCODING) { + let text = value.to_str().map_err(|_| { + rejection( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "only gzip content encoding is accepted".into(), + ) + })?; + encodings.extend( + text.split(',') + .map(|encoding| encoding.trim().to_ascii_lowercase()) + .filter(|encoding| !encoding.is_empty() && encoding != IDENTITY_ENCODING), + ); + } + Ok(encodings) +} + +/// The statuses axum's JSON extractor uses: malformed JSON is 400, a wrong shape is 422. +fn parse_rejection(error: &serde_json::Error) -> ApiError { + if error.is_data() { + rejection( + StatusCode::UNPROCESSABLE_ENTITY, + format!("Failed to deserialize the JSON body into the target type: {error}"), + ) + } else { + rejection( + StatusCode::BAD_REQUEST, + format!("Failed to parse the request body as JSON: {error}"), + ) + } +} + +fn rejection(status: StatusCode, message: String) -> ApiError { + ApiError { + status, + message, + decision_id: None, + } +} diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs index 99afc12f8..c9cb63ced 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs @@ -1,12 +1,15 @@ //! A location credential publishes and reads only its own location's snapshots. +use axum::body::Bytes; use axum::extract::{Path, State}; -use axum::http::StatusCode; +use axum::http::{HeaderMap, StatusCode}; use axum::{Extension, Json}; use uuid::Uuid; use crate::api::_helpers::authentication::bearer::require_location; use crate::api::_helpers::resources::docker_image; +use crate::api::_helpers::resources::request_json::request_json; +use crate::api::constants::DECOMPRESSED_BODY_BYTES; use crate::api::errors::ApiError; use crate::api::models::{DevelopmentScope, DockerImageSource}; use crate::credentials::models::TokenIdentity; @@ -22,20 +25,30 @@ use crate::store::models::ControlStore; request_body = SnapshotInput, responses((status = 200, body = SnapshotRecord), (status = 400, body = crate::api::models::ApiErrorBody), (status = 401, description = "Invalid credential"), (status = 403, description = "Location not authorized"), - (status = 409, description = "Stale generation")), security(("bearer" = [])))] + (status = 409, description = "Stale generation"), (status = 413, description = "Body over its wire or decompressed bound"), + (status = 415, description = "Content encoding other than gzip")), security(("bearer" = [])))] pub(crate) async fn publish( State(store): State, (Extension(identity), Extension(images)): ( Extension, Extension>>, ), - (Path((location_id, target)), development): ( + (Path((location_id, target)), development, headers): ( Path<(Uuid, String)>, Option>, + HeaderMap, ), - Json(input): Json, + body: Bytes, ) -> Result, ApiError> { require_location(&identity, location_id)?; + let input: SnapshotInput = + tokio::task::spawn_blocking(move || request_json(&headers, &body, DECOMPRESSED_BODY_BYTES)) + .await + .map_err(|_| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: "snapshot decoding stopped".into(), + decision_id: None, + })??; if input.location_id != location_id || input.target != target { return Err(ApiError { status: StatusCode::BAD_REQUEST, diff --git a/crates/sqlbuild-orchestration/src/api/constants.rs b/crates/sqlbuild-orchestration/src/api/constants.rs index 9bd3b7c6f..89bb61672 100644 --- a/crates/sqlbuild-orchestration/src/api/constants.rs +++ b/crates/sqlbuild-orchestration/src/api/constants.rs @@ -1,6 +1,11 @@ //! Shared HTTP transport bounds for the server and native clients. pub const BODY_BYTES: usize = 64 * 1024 * 1024; +/// A gzip body may decompress to this many bytes; a parsed snapshot holds about 7.6 times its JSON size. +pub(crate) const DECOMPRESSED_BODY_BYTES: usize = 320 * 1024 * 1024; +pub(crate) const GZIP_ENCODING: &str = "gzip"; +pub(crate) const IDENTITY_ENCODING: &str = "identity"; +pub(crate) const JSON_MEDIA_TYPE: &str = "application/json"; /// Two 1024-byte passwords can each expand sixfold through JSON escaping. pub(crate) const PASSWORD_BODY_BYTES: usize = 16 * 1024; pub(crate) const PASSWORD_COOKIE: &str = "__Host-sqb-session"; diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 9571345a4..f16da08e5 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -431,3 +431,11 @@ pub(crate) struct UiAsset { pub(crate) path: &'static str, pub(crate) body: &'static [u8], } + +/// A decompressing reader that fails once it would yield more than `remaining` bytes. +#[derive(Debug)] +pub(crate) struct BoundedReader { + pub(crate) inner: R, + pub(crate) remaining: usize, + pub(crate) exceeded: bool, +} diff --git a/crates/sqlbuild-orchestration/src/snapshots/main/publish_remote_snapshot.rs b/crates/sqlbuild-orchestration/src/snapshots/main/publish_remote_snapshot.rs index 681cc0f61..f278709c9 100644 --- a/crates/sqlbuild-orchestration/src/snapshots/main/publish_remote_snapshot.rs +++ b/crates/sqlbuild-orchestration/src/snapshots/main/publish_remote_snapshot.rs @@ -1,10 +1,16 @@ //! A code location sends retained compiled data over HTTP and never opens the control store. +use reqwest::StatusCode; +use reqwest::header::{CONTENT_ENCODING, CONTENT_TYPE}; + +use crate::api::constants::{BODY_BYTES, GZIP_ENCODING, JSON_MEDIA_TYPE}; use crate::api::errors::ClientError; use crate::api::main::client_response::client_response; use crate::api::main::connect_client::connect_client; use crate::snapshots::models::{SnapshotInput, SnapshotRecord}; +use crate::store::main::compress::compress; +/// Sends the snapshot gzip-encoded, and once more as plain JSON when a server refuses the encoding and the plain body fits the wire bound. pub async fn publish_remote_snapshot( server_url: &str, bearer: &str, @@ -19,5 +25,25 @@ pub async fn publish_remote_snapshot( &input.target, "snapshots", ])?; - client_response(connection.client.post(url).json(input)).await + let body = serde_json::to_vec(input) + .map_err(|_| ClientError::Configuration("snapshot is not serializable as JSON"))?; + let gzipped = + compress(&body).map_err(|_| ClientError::Configuration("snapshot could not be gzipped"))?; + let request = |body: Vec| { + connection + .client + .post(url.clone()) + .header(CONTENT_TYPE, JSON_MEDIA_TYPE) + .body(body) + }; + match client_response(request(gzipped).header(CONTENT_ENCODING, GZIP_ENCODING)).await { + Err(ClientError::Rejected(status)) + if (status == StatusCode::BAD_REQUEST + || status == StatusCode::UNSUPPORTED_MEDIA_TYPE) + && body.len() <= BODY_BYTES => + { + client_response(request(body)).await + } + result => result, + } } diff --git a/crates/sqlbuild-orchestration/tests/api/helpers.rs b/crates/sqlbuild-orchestration/tests/api/helpers.rs index daa60882a..5b29e670a 100644 --- a/crates/sqlbuild-orchestration/tests/api/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/helpers.rs @@ -101,3 +101,53 @@ pub(crate) fn raw_get(address: &str, path: &str) -> (u16, String) { .expect("numeric status"); (status, body.to_owned()) } + +/// Gzip members that decompress to a JSON string field longer than the decompressed bound. +pub(crate) fn gzip_bomb() -> Vec { + let filler = crate::store::main::compress::compress(&vec![b'a'; 1024 * 1024]).expect("gzip"); + let mut body = crate::store::main::compress::compress(br#"{"padding":""#).expect("gzip"); + for _ in 0..=crate::api::constants::DECOMPRESSED_BODY_BYTES / (1024 * 1024) { + body.extend_from_slice(&filler); + } + body +} + +/// Plain JSON one byte longer than the wire bound. +pub(crate) fn oversized_json() -> Vec { + let mut body = br#"{"padding":""#.to_vec(); + body.resize(crate::api::constants::BODY_BYTES - 1, b'a'); + body.extend_from_slice(br#""}"#); + body +} + +pub(crate) fn gzipped_empty_object() -> Vec { + crate::store::main::compress::compress(b"{}").expect("gzip") +} + +pub(crate) fn not_gzip() -> Vec { + b"{}".to_vec() +} + +/// A DuckShop-10k-sized snapshot whose plain JSON is larger than the wire bound. +pub(crate) fn large_snapshot(location_id: uuid::Uuid) -> crate::snapshots::models::SnapshotInput { + let mut input = crate::runs::tests::helpers::snapshot(location_id); + input.nodes = (0..10_282) + .map(|index| { + let sql = format!( + "select {index} as order_id, '{}' as note from orders_{index}", + "x".repeat(3_400) + ); + ( + format!("model:orders_{index}"), + crate::snapshots::models::SnapshotNode { + kind: "model".into(), + definition: serde_json::json!({"materialization": "table"}), + raw_sql: Some(sql.clone()), + compiled_sql: Some(sql), + python_source: None, + }, + ) + }) + .collect(); + input +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs index b59038211..9265f0c7f 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs @@ -70,6 +70,8 @@ mod test_person_location_work; mod test_person_sessions; #[path = "test_proxy.rs"] mod test_proxy; +#[path = "test_publish_encoding.rs"] +mod test_publish_encoding; #[path = "test_reads.rs"] mod test_reads; #[path = "test_statistics_recovery.rs"] diff --git a/crates/sqlbuild-orchestration/tests/api/test_publish_encoding.rs b/crates/sqlbuild-orchestration/tests/api/test_publish_encoding.rs new file mode 100644 index 000000000..9b2aefbcf --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_publish_encoding.rs @@ -0,0 +1,222 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use axum::http::HeaderMap; +use axum::routing::post; +use axum::{Json, Router}; +use chrono::Utc; +use reqwest::StatusCode; +use uuid::Uuid; + +use crate::api::constants::BODY_BYTES; +use crate::api::errors::ClientError; +use crate::api::tests::helpers::{ + gzip_bomb, gzipped_empty_object, large_snapshot, not_gzip, oversized_json, start_api, stop_api, +}; +use crate::api::tests::test_types::{OlderServerTestCase, PublishEncodingTestCase}; +use crate::credentials::main::tokens::issue_token::issue_token; +use crate::credentials::models::{AuditContext, CredentialScope}; +use crate::credentials::tests::location_helpers::create_location; +use crate::runs::tests::helpers::snapshot; +use crate::snapshots::main::publish_remote_snapshot::publish_remote_snapshot; +use crate::snapshots::models::{SnapshotInput, SnapshotRecord}; +use crate::store::main::migrate_store::migrate_store; +use crate::store::tests::store_helpers::{open_test_store, run}; +use crate::store::tests::store_test_types::Backend; + +/// Decoding happens before the store is touched, so one backend covers every encoding case. +#[test] +fn given_encoded_snapshot_bodies_when_publishing_then_gzip_is_bounded_after_authentication() { + let test_cases = [ + PublishEncodingTestCase { + description: "an unauthenticated gzip bomb is refused before it is decompressed", + forged_bearer: Some("unknown.credential"), + encoding: "gzip", + body: gzip_bomb, + expected_status: StatusCode::UNAUTHORIZED, + }, + PublishEncodingTestCase { + description: "a gzip body that decompresses past the bound is refused", + forged_bearer: None, + encoding: "gzip", + body: gzip_bomb, + expected_status: StatusCode::PAYLOAD_TOO_LARGE, + }, + PublishEncodingTestCase { + description: "a plain body over the wire bound is refused", + forged_bearer: None, + encoding: "identity", + body: oversized_json, + expected_status: StatusCode::PAYLOAD_TOO_LARGE, + }, + PublishEncodingTestCase { + description: "an encoding other than gzip is refused", + forged_bearer: None, + encoding: "br", + body: gzipped_empty_object, + expected_status: StatusCode::UNSUPPORTED_MEDIA_TYPE, + }, + PublishEncodingTestCase { + description: "a body declared gzip that is not gzip is refused", + forged_bearer: None, + encoding: "gzip", + body: not_gzip, + expected_status: StatusCode::BAD_REQUEST, + }, + ]; + run(async { + let Some(fixture) = open_test_store(Backend::Sqlite).await else { + return; + }; + let store = &fixture.store; + migrate_store(store).await.expect("migrate"); + let location_id = create_location(store, "orders", "test-admin") + .await + .expect("location"); + let audit = AuditContext { + actor: "test-admin".into(), + granting_role: "admin".into(), + request_id: Uuid::now_v7(), + ip_address: None, + }; + let token = issue_token( + store, + &audit, + CredentialScope::Location { location_id }, + Utc::now() + chrono::Duration::hours(1), + ) + .await + .expect("location credential"); + let api = start_api(store).await; + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(120)) + .build() + .expect("HTTP client"); + let path = format!( + "{}/api/locations/{location_id}/targets/development/snapshots", + api.url + ); + for test_case in &test_cases { + let bearer: &str = test_case.forged_bearer.unwrap_or(&token.bearer); + let status = client + .post(&path) + .header("Content-Type", "application/json") + .header("Content-Encoding", test_case.encoding) + .bearer_auth(bearer) + .body((test_case.body)()) + .send() + .await + .expect("a response, not a reset connection") + .status(); + assert_eq!( + status, test_case.expected_status, + "{}", + test_case.description + ); + } + let input = large_snapshot(location_id); + assert!( + serde_json::to_vec(&input).expect("JSON").len() > BODY_BYTES, + "the snapshot only fits the wire bound gzip-encoded" + ); + let published = publish_remote_snapshot(&api.url, &token.bearer, &input) + .await + .expect("publish a 10k-node snapshot"); + assert_eq!(published.generation, 1); + let invalid = SnapshotInput { + generation: 0, + ..input + }; + let refused = publish_remote_snapshot(&api.url, &token.bearer, &invalid).await; + assert!( + matches!(refused, Err(ClientError::Rejected(StatusCode::BAD_REQUEST))), + "a snapshot too large to resend plain reports the server's own 400, got {refused:?}" + ); + let current: serde_json::Value = client + .get(&path) + .bearer_auth(&token.bearer) + .send() + .await + .expect("current snapshot") + .json() + .await + .expect("snapshot JSON"); + assert_eq!( + current["nodes"].as_object().map(serde_json::Map::len), + Some(10_282) + ); + stop_api(api).await; + }); +} + +#[test] +fn given_a_server_that_refuses_gzip_when_publishing_then_the_snapshot_is_resent_as_plain_json() { + let test_cases = [ + OlderServerTestCase { + description: "a server that predates gzip rejects the encoded body as malformed JSON", + refusal: StatusCode::BAD_REQUEST, + expected_requests: 2, + }, + OlderServerTestCase { + description: "a server that refuses the encoding", + refusal: StatusCode::UNSUPPORTED_MEDIA_TYPE, + expected_requests: 2, + }, + ]; + for test_case in test_cases { + run(async { + let requests = Arc::new(AtomicUsize::new(0)); + let seen = Arc::clone(&requests); + let refusal = test_case.refusal; + let router = Router::new().route( + "/api/locations/{location_id}/targets/{target}/snapshots", + post(move |headers: HeaderMap, body: axum::body::Bytes| { + let seen = Arc::clone(&seen); + async move { + seen.fetch_add(1, Ordering::SeqCst); + headers + .get("content-encoding") + .map_or(Ok(()), |_| Err(refusal))?; + let input: SnapshotInput = + serde_json::from_slice(&body).map_err(|_| StatusCode::BAD_REQUEST)?; + Ok::<_, StatusCode>(Json(SnapshotRecord { + snapshot_id: Uuid::nil(), + location_id: input.location_id, + target: input.target, + generation: input.generation, + manifest_digest: "older-server".into(), + published_at: Utc::now(), + })) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("older server socket"); + let url = format!("http://{}", listener.local_addr().expect("address")); + let server = tokio::spawn(async move { axum::serve(listener, router).await }); + let location_id = Uuid::now_v7(); + let published = publish_remote_snapshot(&url, "token.secret", &snapshot(location_id)) + .await + .expect("plain JSON publish to an older server"); + assert_eq!( + published.manifest_digest, "older-server", + "{}", + test_case.description + ); + assert_eq!( + published.location_id, location_id, + "{}", + test_case.description + ); + assert_eq!( + requests.load(Ordering::SeqCst), + test_case.expected_requests, + "{}", + test_case.description + ); + server.abort(); + }); + } +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_types.rs b/crates/sqlbuild-orchestration/tests/api/test_types.rs index 5d3d2b50f..7f73e0b6c 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_types.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_types.rs @@ -8,6 +8,20 @@ pub(crate) struct ApiContractTestCase { pub(crate) expected_status: RunStatus, } +pub(crate) struct PublishEncodingTestCase { + pub(crate) description: &'static str, + pub(crate) forged_bearer: Option<&'static str>, + pub(crate) encoding: &'static str, + pub(crate) body: fn() -> Vec, + pub(crate) expected_status: reqwest::StatusCode, +} + +pub(crate) struct OlderServerTestCase { + pub(crate) description: &'static str, + pub(crate) refusal: reqwest::StatusCode, + pub(crate) expected_requests: usize, +} + pub(crate) struct DevelopmentTestCase { pub(crate) description: &'static str, pub(crate) backend: Backend,