fix: answer unreadable request formats with 415, and count content types (DW-985)

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::<LengthLimitError>().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<P: PoolProvider>(
     current_user: RequiresPermission<resource::Files, operation::CreateOwn>,
     request: axum::http::Request<axum::body::Body>,
 ) -> Result<(StatusCode, Json<FileResponse>)> {
+    // 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) {

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 { .. } => {

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<Arc<Semaphore>> =
 /// 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<usize>, request: Request<Body>, 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 {

onwards/src/strict/handlers.rs

@@ -409,7 +409,7 @@ pub async fn responses_handler<T: HttpClient + Clone + Send + Sync + 'static>(
         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<T: HttpClient + Clone + Send + Sync + 'static>(
                         &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);
         }
     };
 

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")]),