diff --git a/docs/rust-server-port-checkpoints/2026-10-22-payload-limits.md b/docs/rust-server-port-checkpoints/2026-10-22-payload-limits.md new file mode 100644 index 000000000..b9f32a869 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-22-payload-limits.md @@ -0,0 +1,32 @@ +# October 22, request body admission + +- **Gap.** Published PHP prepends `EnforcePayloadLimits` to every request, + ahead of authentication and routing. Native checked malformed JSON only on + the two system passes. Elsewhere it answered a plain-text 400 or 415 from + its JSON extractor, after authentication. Its body cap was a fixed 3 MB. +- **Source.** A new outermost layer applies PHP's checks in PHP's order: + - **Size.** A declared `Content-Length` or a body over + `server.limits.max_payload_bytes` (`DW_MAX_PAYLOAD_BYTES` or + `WORKFLOW_MAX_PAYLOAD_BYTES`, default 2 MiB) is refused with + `413 payload_too_large` and the limit. + - **Media type.** A POST, PUT, PATCH or DELETE with a body must use + `application/json` or a `+json` type, or it is refused with + `415 unsupported_media_type` and the accepted types. + - **JSON.** A body PHP's `json_decode` would refuse is refused with + `400 malformed_json` and PHP's `json_last_error_msg()` text: syntax, + malformed UTF-8, control character, unpaired surrogate, or nesting deeper + than 512 levels. A whitespace-only body is malformed, as in PHP. + - **Envelope.** Worker-plane routes answer in the worker protocol envelope. + Others use the control-plane envelope. + - **Exception.** The external payload upload keeps its own transport rules. + The layer reads each body once and bounds it, so the fixed 3 MB cap is + removed. +- **Remaining gap.** A body nested 129 to 512 levels deep passes this check, + as in PHP, but native's handlers then refuse it (the port's gap list). +- **Verification.** + - Unit tests map serde_json's errors to PHP's messages, including both + unpaired surrogate forms, a valid surrogate pair and the 512-level depth. + - A new integration test checks unauthenticated malformed bodies, worker + and control-plane envelopes, a missing or non-JSON media type, an + oversized body and a `+json` type passing on to authentication. + - The full SQLite suite passes. diff --git a/docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md b/docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md new file mode 100644 index 000000000..f202466fa --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md @@ -0,0 +1,26 @@ +# October 22, query worker results + +- **Found by.** The Shared Server fixtures workflow on main. After #626, its + `mysql-execution` and `postgres-storage` jobs still failed in the SDK query + directions profile (main run 39704552600). The PHP-worker cases differed in + `control_plane.result`: + - native's wire response had `null`; + - the published Python client's hydrated response had the query state. +- **Cause.** Published PHP SDK workers complete a query task with both the + plain `result` and its `result_envelope`. Published PHP Server's + `completeQueryTask` accepts both and stores both. Its answer returns the + plain `result` beside the envelope, and the control-plane envelope copies + it. Native required the envelope, ignored the plain `result`, and always + answered `result: null`. +- **Source.** + - **Storage.** A completed query task keeps the worker's plain `result`. + - **Answer.** The answer returns it, and its control-plane envelope copies + it, as PHP does. + - **Plain-only completions.** A completion with a plain `result` and no + envelope is accepted, as PHP's rules allow. A completion with neither is + still refused. + - **Budget.** The plain result counts toward the query result budget. +- **Verification.** A new integration test completes one query with a plain + result and an envelope, and one with only a plain result. Both answers + carry the plain result at the top level and in `control_plane`. The query + tests pass on SQLite, PostgreSQL and MySQL. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index b45d6acc1..9175175e2 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -167,10 +167,10 @@ contract fixture before its correction. its message stream items. Native does not clean either up yet. It skips such a run with `external_payload_cleanup_unavailable` or `message_stream_cleanup_unavailable` and keeps its details. -- **Malformed JSON bodies.** Published PHP answers a malformed JSON body - with `400 malformed_json` and a JSON body, and adds the control-plane - envelope on control-plane operations. Native does so only on the two - system passes. Elsewhere, native answers a plain-text 400. +- **Deeply nested JSON bodies.** Published PHP decodes JSON up to 512 levels + deep. Native checks every body as PHP does before any route, but its + handlers parse with serde_json's 128-level limit. A body nested 129 to 512 + levels deep passes the check and is then refused with a plain-text 400. - **Registration without a capability manifest.** At protocol 1.19, published PHP 2.5.14 refuses a registration that has no `capability_manifest` (422). Native accepts it. diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 8427fe452..4137d1196 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -750,20 +750,25 @@ impl Runtime { if let Some(receipt) = receipt { return Ok(receipt); } - let result = super::envelope(&body, "result_envelope")?; - if result.len() > super::queries::RESULT_BYTES { - return Err(refuse( - StatusCode::PAYLOAD_TOO_LARGE, - "query_result_exceeds_budget", - )); + // A plain `result` without an envelope needs no decoding. + let plain_only = body.get("result").is_some_and(|value| !value.is_null()) + && body.get("result_envelope").is_none_or(Value::is_null); + if !plain_only { + let result = super::envelope(&body, "result_envelope")?; + if result.len() > super::queries::RESULT_BYTES { + return Err(refuse( + StatusCode::PAYLOAD_TOO_LARGE, + "query_result_exceeds_budget", + )); + } + super::queries::decode( + self.signal_codec.clone(), + result, + false, + "invalid_query_result", + ) + .await?; } - super::queries::decode( - self.signal_codec.clone(), - result, - false, - "invalid_query_result", - ) - .await?; self.finish_query(task_id, body, false).await } pub(crate) async fn finish_query( diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index b883a7708..000f57598 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -271,8 +271,10 @@ pub fn router(runtime: Runtime) -> Router { .route("/api/worker/query-tasks/{task_id}/fail", post(fail_query)) .fallback(unavailable) .layer(middleware::from_fn(empty_list_body)) - .layer(DefaultBodyLimit::max(3 * 1024 * 1024)) + // `payload_limits` reads and bounds every body first. + .layer(DefaultBodyLimit::disable()) .layer(middleware::from_fn_with_state(runtime.clone(), guard)) + .layer(middleware::from_fn(super::payload_limits::enforce)) .with_state(runtime) } @@ -758,7 +760,7 @@ async fn delete_search_attribute( Ok(Json(runtime.delete_search_attribute(&name).await?)) } -async fn admission_error( +pub(super) async fn admission_error( request: Request, worker: bool, status: StatusCode, diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 5df2948f1..11f7de294 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -42,6 +42,7 @@ pub mod mysql; mod namespaces; mod operator_dashboard; mod operator_metrics; +mod payload_limits; mod poll_readiness; pub mod postgres; mod previews; diff --git a/rust/src/runtime/payload_limits.rs b/rust/src/runtime/payload_limits.rs new file mode 100644 index 000000000..81a273568 --- /dev/null +++ b/rust/src/runtime/payload_limits.rs @@ -0,0 +1,152 @@ +//! Published PHP Server's `EnforcePayloadLimits`, which runs before every +//! other middleware, authentication included: a body over the payload limit, +//! a body without a JSON media type and a body that is not valid JSON are each +//! refused with PHP's JSON error, in the worker or control-plane envelope. +//! The external payload upload keeps its own transport rules. +use axum::{ + body::{Body, to_bytes}, + extract::Request, + http::{Method, StatusCode, header}, + middleware::Next, + response::Response, +}; +use serde_json::{Value, json}; + +/// `server.limits.max_payload_bytes`. +fn max_payload_bytes() -> usize { + ["DW_MAX_PAYLOAD_BYTES", "WORKFLOW_MAX_PAYLOAD_BYTES"] + .iter() + .find_map(|key| std::env::var(key).ok()) + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(2 * 1024 * 1024) +} + +pub(super) async fn enforce(request: Request, next: Next) -> Response { + let path = request.uri().path().to_owned(); + if request.method() == Method::POST && path == "/api/external-payloads/v1" { + return next.run(request).await; + } + let max = max_payload_bytes(); + let worker = path == "/api/worker" || path.starts_with("/api/worker/"); + let declared = request + .headers() + .get(header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.trim().parse::().ok()); + let (parts, body) = request.into_parts(); + let refuse = |status: StatusCode, body: Value| { + let parts = parts.clone(); + async move { super::http::admission_error(Request::from_parts(parts, Body::empty()), worker, status, body).await } + }; + let too_large = || json!({"message": format!("Request payload exceeds the maximum allowed size of {max} bytes."), + "reason": "payload_too_large", "limit": max}); + if declared.is_some_and(|declared| declared > max as u64) { + return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await; + } + let Ok(bytes) = to_bytes(body, max.saturating_add(1)).await else { + return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await; + }; + if bytes.len() > max { + return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await; + } + let has_body = !bytes.is_empty() || declared.is_some_and(|declared| declared > 0); + if matches!(parts.method, Method::POST | Method::PUT | Method::PATCH | Method::DELETE) && has_body { + if !json_media_type(parts.headers.get(header::CONTENT_TYPE).and_then(|value| value.to_str().ok())) { + return refuse(StatusCode::UNSUPPORTED_MEDIA_TYPE, json!({"message": "Request bodies must use a JSON media type.", + "reason": "unsupported_media_type", "accepted_content_types": ["application/json", "application/*+json"]})).await; + } + if let Some(error) = json_error(&bytes) { + return refuse(StatusCode::BAD_REQUEST, json!({"message": "Request bodies must contain valid JSON.", + "reason": "malformed_json", "json_error": error})).await; + } + } + next.run(Request::from_parts(parts, Body::from(bytes))).await +} + +/// `usesJsonMediaType`: `application/json` or any `+json` type. +fn json_media_type(content_type: Option<&str>) -> bool { + let Some(content_type) = content_type.filter(|value| !value.trim().is_empty()) else { + return false; + }; + let media = content_type.split(';').next().unwrap_or("").trim().to_ascii_lowercase(); + media == "application/json" || media.ends_with("+json") +} + +/// PHP's `json_last_error_msg()` after `json_decode` of the body, or `None` +/// for valid JSON. PHP nests up to 512 levels. +fn json_error(bytes: &[u8]) -> Option<&'static str> { + // Skipping a value leaves escapes and UTF-8 unchecked, so decode it. + let error = serde_json::from_slice::(bytes).err()?; + let message = error.to_string(); + Some(if message.contains("recursion limit exceeded") { + if depth(bytes) <= 512 { + return None; + } + "Maximum stack depth exceeded" + } else if message.contains("control character") { + "Control character error, possibly incorrectly encoded" + } else if message.contains("surrogate") || message.contains("unexpected end of hex escape") { + "Single unpaired UTF-16 surrogate in unicode escape" + } else if message.contains("invalid unicode code point") { + "Malformed UTF-8 characters, possibly incorrectly encoded" + } else { + "Syntax error" + }) +} + +/// The deepest array or object nesting, outside strings. +fn depth(bytes: &[u8]) -> usize { + let (mut current, mut deepest, mut in_string, mut escaped) = (0usize, 0usize, false, false); + for &byte in bytes { + if in_string { + match (escaped, byte) { + (true, _) => escaped = false, + (false, b'\\') => escaped = true, + (false, b'"') => in_string = false, + _ => {} + } + continue; + } + match byte { + b'"' => in_string = true, + b'[' | b'{' => { + current += 1; + deepest = deepest.max(current); + } + b']' | b'}' => current = current.saturating_sub(1), + _ => {} + } + } + deepest +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn errors_name_php_s_json_failures() { + assert_eq!(json_error(br#"{"a":1}"#), None); + assert_eq!(json_error(b"\"text\""), None); + assert_eq!(json_error(br#"{"a":"#), Some("Syntax error")); + assert_eq!(json_error(b" "), Some("Syntax error")); + assert_eq!(json_error(b"{\"a\":\"\x01\"}"), Some("Control character error, possibly incorrectly encoded")); + assert_eq!(json_error(br#"{"a":"\ud800"}"#), Some("Single unpaired UTF-16 surrogate in unicode escape")); + assert_eq!(json_error(br#"{"a":"\udc00"}"#), Some("Single unpaired UTF-16 surrogate in unicode escape")); + assert_eq!(json_error(br#"{"a":"\ud83d\ude00"}"#), None); + assert_eq!(json_error(b"{\"a\":\"\xff\"}"), Some("Malformed UTF-8 characters, possibly incorrectly encoded")); + // PHP accepts nesting serde_json's default limit refuses. + let nested = |levels: usize| format!("{}{}", "[".repeat(levels), "]".repeat(levels)).into_bytes(); + assert_eq!(json_error(&nested(200)), None); + assert_eq!(json_error(&nested(600)), Some("Maximum stack depth exceeded")); + } + + #[test] + fn media_types_follow_php() { + assert!(json_media_type(Some("application/json"))); + assert!(json_media_type(Some("Application/JSON; charset=utf-8"))); + assert!(json_media_type(Some("application/vnd.api+json"))); + assert!(!json_media_type(Some("text/plain"))); + assert!(!json_media_type(None)); + } +} diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 52a9b5bf3..b2245e017 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -458,7 +458,7 @@ where match task["status"].as_str() { Some("completed") => { let mut answer = json!({"workflow_id":workflow_id,"run_id":run_id,"target_scope":"instance", - "query_name":name,"result":null,"result_envelope":task["result_envelope"],"reason":null}); + "query_name":name,"result":task["result"],"result_envelope":task["result_envelope"],"reason":null}); if task["principal"].is_object() { answer["principal"] = task["principal"].clone(); } @@ -1153,6 +1153,7 @@ where let reason: Option; let failure; let result; + let mut plain = None; if failed { let supplied = &body["failure"]; if !supplied.is_object() || serde_json::to_vec(supplied)?.len() > 32 * 1024 { @@ -1171,14 +1172,22 @@ where failure = Some(supplied.clone()); result = None; } else { - let supplied = envelope(body, "result_envelope")?; - if supplied.len() > RESULT_BYTES { + // `completeQueryTask`: a worker may also send the plain `result`, + // which published PHP returns to the caller beside the envelope. + plain = body.get("result").filter(|value| !value.is_null()).cloned(); + let supplied = if plain.is_none() || body.get("result_envelope").is_some_and(|value| !value.is_null()) { + Some(envelope(body, "result_envelope")?) + } else { + None + }; + let plain_bytes = plain.as_ref().map(serde_json::to_vec).transpose()?.map_or(0, |bytes| bytes.len()); + if supplied.as_ref().map_or(0, String::len) + plain_bytes > RESULT_BYTES { return Err(refuse( StatusCode::PAYLOAD_TOO_LARGE, "query_result_exceeds_budget", )); } - result = Some(wire(&supplied)); + result = supplied.map(|supplied| wire(&supplied)); failure = None; reason = None; } @@ -1196,6 +1205,9 @@ where if let Some(result) = result { task["result_envelope"] = result; } + if let Some(plain) = plain { + task["result"] = plain; + } task["terminal_receipt"] = json!({"lease_owner":owner,"attempt":attempt, "fingerprint":fingerprint,"decision":decision,"decided_at":authority::wire(decided_at)}); // Results reserve their slot at admission, and discard the large replay diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 46752e475..3bb0a3a82 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -79,6 +79,10 @@ mod worker_deregistration; mod worker_listing; #[path = "execution/activity_failure_fields.rs"] mod activity_failure_fields; +#[path = "execution/query_worker_results.rs"] +mod query_worker_results; +#[path = "execution/payload_limits.rs"] +mod payload_limits; #[path = "execution/query_open_tasks.rs"] mod query_open_tasks; #[path = "execution/retry_message_tasks.rs"] diff --git a/rust/tests/execution/payload_limits.rs b/rust/tests/execution/payload_limits.rs new file mode 100644 index 000000000..17b53a48c --- /dev/null +++ b/rust/tests/execution/payload_limits.rs @@ -0,0 +1,66 @@ +use super::*; + +/// Sends `body` raw, with an optional content type and without credentials +/// unless `authorized`, and returns the status, headers and JSON answer. +async fn raw( + app: &Router, + uri: &str, + content_type: Option<&str>, + body: Vec, + authorized: bool, +) -> (StatusCode, axum::http::HeaderMap, Value) { + let mut request = Request::builder().method("POST").uri(uri).header("content-length", body.len()); + if let Some(content_type) = content_type { + request = request.header("content-type", content_type); + } + if authorized { + request = request.header("authorization", "Bearer test-token"); + } + let response = app.clone().oneshot(request.body(Body::from(body)).unwrap()).await.unwrap(); + let (status, headers) = (response.status(), response.headers().clone()); + let body = to_bytes(response.into_body(), 1 << 20).await.unwrap(); + (status, headers, serde_json::from_slice(&body).unwrap_or_else(|_| panic!("{body:?}"))) +} + +// Published PHP's EnforcePayloadLimits runs before authentication and every +// route: oversized bodies, non-JSON media types and malformed JSON are +// refused in the worker or control-plane envelope. +#[tokio::test] +async fn bodies_are_checked_before_authentication_like_php() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let json = Some("application/json"); + + for (body, error) in [ + (b"{".to_vec(), "Syntax error"), + (b" ".to_vec(), "Syntax error"), + (b"{\"a\":\"\xff\"}".to_vec(), "Malformed UTF-8 characters, possibly incorrectly encoded"), + (b"{\"a\":\"\x01\"}".to_vec(), "Control character error, possibly incorrectly encoded"), + ] { + let (status, headers, answer) = raw(&app, "/api/workflows", json, body, false).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{answer}"); + assert_eq!(headers["x-durable-workflow-control-plane-version"], "2"); + assert_eq!(answer["message"], "Request bodies must contain valid JSON."); + assert_eq!((answer["reason"].clone(), answer["json_error"].clone()), (json!("malformed_json"), json!(error))); + } + + let (status, headers, answer) = raw(&app, "/api/worker/register", Some("text/plain"), b"{}".to_vec(), false).await; + assert_eq!(status, StatusCode::UNSUPPORTED_MEDIA_TYPE, "{answer}"); + assert!(headers.contains_key("x-durable-workflow-protocol-version")); + assert_eq!(answer["reason"], "unsupported_media_type"); + assert_eq!(answer["accepted_content_types"], json!(["application/json", "application/*+json"])); + assert!(answer["protocol_version"].is_string() && answer["server_capabilities"].is_object(), "{answer}"); + let (status, _, answer) = raw(&app, "/api/worker/register", None, b"{}".to_vec(), true).await; + assert_eq!((status, answer["reason"].clone()), (StatusCode::UNSUPPORTED_MEDIA_TYPE, json!("unsupported_media_type"))); + + let limit = 2 * 1024 * 1024; + let oversized = format!("{{\"input\":\"{}\"}}", "x".repeat(limit)).into_bytes(); + let (status, _, answer) = raw(&app, "/api/workflows", json, oversized, true).await; + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE, "{answer}"); + assert_eq!(answer["message"], format!("Request payload exceeds the maximum allowed size of {limit} bytes.")); + assert_eq!((answer["reason"].clone(), answer["limit"].clone()), (json!("payload_too_large"), json!(limit))); + + // A `+json` media type passes on to the route, which still authenticates. + let (status, _, answer) = raw(&app, "/api/workflows", Some("application/vnd.api+json"), b"{}".to_vec(), false).await; + assert_eq!(status, StatusCode::UNAUTHORIZED, "{answer}"); +} diff --git a/rust/tests/execution/query_worker_results.rs b/rust/tests/execution/query_worker_results.rs new file mode 100644 index 000000000..14ae7ec86 --- /dev/null +++ b/rust/tests/execution/query_worker_results.rs @@ -0,0 +1,40 @@ +//! A query worker's plain `result`, which published PHP stores beside the +//! result envelope and returns to the caller (`completeQueryTask`). +use super::*; + +async fn ask(app: &Router, workflow_id: &str) -> tokio::task::JoinHandle<(StatusCode, Value)> { + let app = app.clone(); + let path = format!("/api/workflows/{workflow_id}/query/state"); + tokio::spawn(async move { + request(&app, "POST", &path, json!({"input":envelope(Payload::Array(vec![Payload::Long(1)]))})).await + }) +} + +#[tokio::test] +async fn query_answers_carry_the_worker_s_plain_result() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + query_worker(&app, "plain-query-worker").await; + let plain = json!({"received":0,"last":null}); + for (workflow_id, with_envelope) in [("plain-and-envelope", true), ("plain-only", false)] { + start(&app, workflow_id).await; + let asked = ask(&app, workflow_id).await; + let claim = await_query(&app, "plain-query-worker").await; + let mut completion = json!({"lease_owner":"plain-query-worker","query_task_attempt":claim["query_task_attempt"], + "result":plain}); + if with_envelope { + completion["result_envelope"] = envelope(Payload::Long(7)); + } + let (status, body) = request(&app, "POST", + &format!("/api/worker/query-tasks/{}/complete", claim["query_task_id"].as_str().unwrap()), completion).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, answer) = asked.await.unwrap(); + assert_eq!(status, StatusCode::OK, "{answer}"); + assert_eq!(answer["result"], plain, "{answer}"); + assert_eq!(answer["control_plane"]["result"], plain, "{answer}"); + assert_eq!(answer["result_envelope"].is_object(), with_envelope, "{answer}"); + } + runtime.close().await; + database.remove().await; +}