diff --git a/dwctl/src/api/handlers/files.rs b/dwctl/src/api/handlers/files.rs index 0ec11fe36..05cc3f081 100644 --- a/dwctl/src/api/handlers/files.rs +++ b/dwctl/src/api/handlers/files.rs @@ -584,7 +584,7 @@ impl FileUploadError { FileUploadError::InvalidCompression { format, message } => Error::BadRequest { message: format!("File is not valid {format} data: {message}"), }, - FileUploadError::UnsupportedCompression { format } => Error::BadRequest { + FileUploadError::UnsupportedCompression { format } => Error::UnsupportedMediaType { message: format!( "File is compressed with {format}, which is not supported. Upload it uncompressed, or compressed with gzip or zstd." ), @@ -635,6 +635,15 @@ impl FileUploadError { /// Check if an error (or any error in its source chain) is a body/stream length limit error. /// Uses typed error matching via downcasting, with string-based fallback for wrapped errors. +/// Whether the request declares a `multipart/form-data` body, parameters aside. +fn is_multipart_form(headers: &axum::http::HeaderMap) -> bool { + headers + .get(axum::http::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|mime| mime.trim().eq_ignore_ascii_case("multipart/form-data")) +} + fn is_length_limit_error(err: &(dyn std::error::Error + 'static)) -> bool { // Check for axum's LengthLimitError directly if err.downcast_ref::().is_some() { @@ -1169,6 +1178,7 @@ Each line must be a valid JSON object containing `custom_id`, `method`, `url`, a (status = 400, description = "Invalid file format, malformed JSON, missing required fields, etc."), (status = 403, description = "Model referenced in the file is not configured or not accessible to your account."), (status = 413, description = "File exceeds the maximum allowed size."), + (status = 415, description = "The request is not `multipart/form-data`, or the file is compressed in a format other than gzip or zstd."), (status = 422, description = "A referenced image URL could not be retrieved because its origin refused access (e.g. 403/404). The file is well-formed but references an image that is forbidden, gated, or missing."), (status = 429, description = "Too many concurrent uploads. Retry after a short delay."), (status = 500, description = "An unexpected error occurred. Retry the request or contact support if the issue persists."), @@ -1181,6 +1191,14 @@ pub async fn upload_file( current_user: RequiresPermission, request: axum::http::Request, ) -> Result<(StatusCode, Json)> { + // Refuse a body that isn't a multipart form before it takes an upload slot. + // A malformed form (a bad boundary, say) is still a 400 below. + if !is_multipart_form(request.headers()) { + return Err(Error::UnsupportedMediaType { + message: "File uploads must be sent as multipart/form-data, with the file in a `file` field".to_string(), + }); + } + // Acquire upload permit (if limiter is configured) // The permit is held for the duration of the upload to limit concurrency let _permit = if let Some(ref limiter) = state.limiters.file_uploads { @@ -2497,7 +2515,10 @@ mod tests { let truncated_gzip = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03\xab\x56".to_vec(); let xz = b"\xfd7zXZ\x00\x00\x04\xe6\xd6\xb4\x46".to_vec(); - for (content, expected) in [(truncated_gzip, "not valid gzip data"), (xz, "compressed with xz")] { + for (content, status, expected) in [ + (truncated_gzip, axum::http::StatusCode::BAD_REQUEST, "not valid gzip data"), + (xz, axum::http::StatusCode::UNSUPPORTED_MEDIA_TYPE, "compressed with xz"), + ] { let upload_response = app .post("/ai/v1/files") .add_header(&add_auth_headers(&user)[0].0, &add_auth_headers(&user)[0].1) @@ -2509,12 +2530,44 @@ mod tests { ) .await; - upload_response.assert_status(axum::http::StatusCode::BAD_REQUEST); + upload_response.assert_status(status); let body = upload_response.text(); assert!(body.contains(expected), "{body}"); } } + #[dwctl_test_macros::test] + #[test_log::test] + async fn test_upload_that_is_not_a_multipart_form_returns_415(pool: PgPool) { + let (app, _bg_services) = create_test_app(pool.clone(), false).await; + let user = create_test_user_with_roles(&pool, vec![Role::StandardUser, Role::BatchAPIUser]).await; + let line = r#"{"custom_id":"request-1","method":"POST","url":"/v1/chat/completions","body":{"model":"gpt-4"}}"#; + + for content_type in [Some("application/jsonl"), Some("application/x-www-form-urlencoded"), None] { + let mut request = app + .post("/ai/v1/files") + .add_header(&add_auth_headers(&user)[0].0, &add_auth_headers(&user)[0].1) + .add_header(&add_auth_headers(&user)[1].0, &add_auth_headers(&user)[1].1); + if let Some(content_type) = content_type { + request = request.content_type(content_type); + } + let upload_response = request.bytes(line.as_bytes().to_vec().into()).await; + + upload_response.assert_status(axum::http::StatusCode::UNSUPPORTED_MEDIA_TYPE); + assert!(upload_response.text().contains("multipart/form-data"), "{content_type:?}"); + } + + // A multipart content type with a broken form is still malformed, not unsupported. + let upload_response = app + .post("/ai/v1/files") + .add_header(&add_auth_headers(&user)[0].0, &add_auth_headers(&user)[0].1) + .add_header(&add_auth_headers(&user)[1].0, &add_auth_headers(&user)[1].1) + .content_type("multipart/form-data") + .bytes(line.as_bytes().to_vec().into()) + .await; + upload_response.assert_status(axum::http::StatusCode::BAD_REQUEST); + } + #[dwctl_test_macros::test] #[test_log::test] async fn test_upload_missing_model_field(pool: PgPool) { diff --git a/dwctl/src/errors.rs b/dwctl/src/errors.rs index 1ad39adc3..ba043312a 100644 --- a/dwctl/src/errors.rs +++ b/dwctl/src/errors.rs @@ -141,6 +141,10 @@ pub enum Error { #[error("Payload too large: {message}")] PayloadTooLarge { message: String }, + /// The request body is in a format or content type the endpoint doesn't read + #[error("Unsupported media type: {message}")] + UnsupportedMediaType { message: String }, + /// Insufficient credits to perform the requested operation #[error("Insufficient credits: {message}")] InsufficientCredits { current_balance: Decimal, message: String }, @@ -209,6 +213,7 @@ impl Error { Error::Other(_) => StatusCode::INTERNAL_SERVER_ERROR, Error::Conflict { .. } => StatusCode::CONFLICT, Error::PayloadTooLarge { .. } => StatusCode::PAYLOAD_TOO_LARGE, + Error::UnsupportedMediaType { .. } => StatusCode::UNSUPPORTED_MEDIA_TYPE, Error::InsufficientCredits { .. } => StatusCode::PAYMENT_REQUIRED, Error::SpendCapExceeded { .. } => StatusCode::PAYMENT_REQUIRED, Error::ModelAccessDenied { .. } => StatusCode::FORBIDDEN, @@ -229,6 +234,7 @@ impl Error { Error::BadRequest { message } => message.clone(), Error::UnprocessableEntity { message } => message.clone(), Error::PayloadTooLarge { message } => message.clone(), + Error::UnsupportedMediaType { message } => message.clone(), Error::NotFound { resource, id } => { format!("{resource} with ID {id} not found") } @@ -304,7 +310,8 @@ impl IntoResponse for Error { | Error::UnprocessableEntity { .. } | Error::NotFound { .. } | Error::Gone { .. } - | Error::PayloadTooLarge { .. } => { + | Error::PayloadTooLarge { .. } + | Error::UnsupportedMediaType { .. } => { tracing::debug!("Client error: {}", self); } Error::Conflict { .. } => { diff --git a/dwctl/src/inference/body_limit.rs b/dwctl/src/inference/body_limit.rs index 94b72f366..71b0e9d65 100644 --- a/dwctl/src/inference/body_limit.rs +++ b/dwctl/src/inference/body_limit.rs @@ -11,8 +11,8 @@ use axum::{ body::{Body, Bytes, to_bytes}, extract::State, http::{ - HeaderMap, HeaderValue, Request, StatusCode, - header::{ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_LENGTH}, + HeaderMap, HeaderValue, Method, Request, StatusCode, + header::{ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TYPE}, }, middleware::Next, response::{IntoResponse, Response}, @@ -51,6 +51,10 @@ static DECODE_SLOTS: LazyLock> = /// body under a wrong or meaningless `Content-Encoding` worked. Such a body /// still does: it is passed on as it is and counted, rather than refused. pub async fn limit_inference_body(State(max_bytes): State, request: Request, next: Next) -> Response { + if request.method() == Method::POST { + metrics::counter!("dwctl_inference_request_content_types_total", "content_type" => content_type_class(request.headers())) + .increment(1); + } let declared = content_encoding(request.headers()); if max_bytes == 0 && declared == Ok(None) { return next.run(request).await; @@ -192,6 +196,30 @@ fn unsupported_encoding(path: &str) -> Response { response } +/// The declared `Content-Type` of an inference body, as a bounded metric +/// label. The JSON endpoints parse the body whatever it declares (only strict +/// `/responses` checks), and clients such as `curl -d` declare a form body; +/// this counts how many do before anything starts refusing them. +fn content_type_class(headers: &HeaderMap) -> &'static str { + let Some(value) = headers.get(CONTENT_TYPE) else { + return "none"; + }; + let mime = value + .to_str() + .ok() + .and_then(|value| value.split(';').next()) + .map(|mime| mime.trim().to_ascii_lowercase()) + .unwrap_or_default(); + match mime.as_str() { + "application/json" => "json", + mime if mime.ends_with("+json") => "json", + "application/x-www-form-urlencoded" => "form", + mime if mime.starts_with("multipart/") => "multipart", + mime if mime.starts_with("text/") => "text", + _ => "other", + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Encoding { Gzip, @@ -765,6 +793,25 @@ mod tests { assert_eq!(content_encoding(&headers(&["deflate"])), Err(UnsupportedEncoding)); } + #[test] + fn classifies_declared_content_types() { + let class = |value: Option<&str>| { + let mut headers = HeaderMap::new(); + if let Some(value) = value { + headers.insert(CONTENT_TYPE, value.parse().unwrap()); + } + content_type_class(&headers) + }; + assert_eq!(class(None), "none"); + assert_eq!(class(Some("application/json")), "json"); + assert_eq!(class(Some("Application/JSON; charset=utf-8")), "json"); + assert_eq!(class(Some("application/vnd.api+json")), "json"); + assert_eq!(class(Some("application/x-www-form-urlencoded")), "form"); + assert_eq!(class(Some("multipart/form-data; boundary=x")), "multipart"); + assert_eq!(class(Some("text/plain")), "text"); + assert_eq!(class(Some("application/octet-stream")), "other"); + } + #[tokio::test] async fn distinguishes_transport_errors_without_exposing_their_contents() { let body = Body::from_stream(futures::stream::once(async { diff --git a/onwards/src/strict/handlers.rs b/onwards/src/strict/handlers.rs index f59276d39..14736e3b6 100644 --- a/onwards/src/strict/handlers.rs +++ b/onwards/src/strict/handlers.rs @@ -409,7 +409,7 @@ pub async fn responses_handler( Err(e) => { let model = extract_model_from_request(&headers, &body_bytes); let message = format!("Invalid request: {e}"); - let code = match e { + let (status, code) = match e { JsonRejection::JsonDataError(_) => { return refused_mismatch( &state, @@ -420,17 +420,14 @@ pub async fn responses_handler( &message, ); } - JsonRejection::MissingJsonContentType(_) => "invalid_content_type", - _ => "invalid_json", + // RFC 9110 ยง15.5.16: a body not declared as JSON is an + // unsupported media type, not a malformed request. + JsonRejection::MissingJsonContentType(_) => { + (StatusCode::UNSUPPORTED_MEDIA_TYPE, "invalid_content_type") + } + _ => (StatusCode::BAD_REQUEST, "invalid_json"), }; - return refused_invalid( - &state, - &headers, - model.as_deref(), - StatusCode::BAD_REQUEST, - code, - &message, - ); + return refused_invalid(&state, &headers, model.as_deref(), status, code, &message); } }; diff --git a/onwards/tests/rejections.rs b/onwards/tests/rejections.rs index bee78b58e..1e1160f75 100644 --- a/onwards/tests/rejections.rs +++ b/onwards/tests/rejections.rs @@ -191,7 +191,7 @@ async fn responses_bodies_not_sent_as_json_are_counted_as_such() { .text(json!({"model": alias, "input": "Hello"}).to_string()) .await; - response.assert_status(StatusCode::BAD_REQUEST); + response.assert_status(StatusCode::UNSUPPORTED_MEDIA_TYPE); assert!(mock.get_requests().is_empty()); assert_eq!( count(&[("model", alias), ("code", "invalid_content_type")]),