diff --git a/Cargo.lock b/Cargo.lock index c4050a9f..ca284661 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2528,12 +2528,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls 0.26.4", + "tokio-util", "tower 0.5.3", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] @@ -3947,6 +3949,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" version = "0.3.103" @@ -3996,6 +4011,7 @@ dependencies = [ "codespan-reporting", "criterion", "dhat", + "encoding_rs", "futures-util", "glob", "hkdf 0.12.4", diff --git a/Cargo.toml b/Cargo.toml index b5391e8d..f74b15ac 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,7 +57,8 @@ regex = "1.13.0" log = "0.4.33" rustyline = "18.0.1" tokio = { version = "1.52.3", features = ["full"] } -reqwest = { version = "0.13.4", features = ["json"] } +reqwest = { version = "0.13.4", features = ["json", "stream"] } +encoding_rs = "0.8.35" # sqlx 0.9 split the old `runtime-tokio-rustls` feature into a separate runtime # and TLS backend; `tls-rustls` aliases the ring-backed rustls stack we used before. sqlx = { version = "0.9.0", features = ["runtime-tokio", "tls-rustls", "sqlite", "mysql", "postgres", "chrono"] } diff --git a/Docs/04-advanced-features/interoperability.md b/Docs/04-advanced-features/interoperability.md index b76bd9a6..a1abd312 100644 --- a/Docs/04-advanced-features/interoperability.md +++ b/Docs/04-advanced-features/interoperability.md @@ -92,6 +92,15 @@ Non-2xx statuses are not errors — check `resp.ok` or `resp.status` yourself. Network failures (DNS, connection refused) still raise errors you can `try`/`catch`. +Outbound responses are streamed and decoded into a bounded buffer. The +`web_server_max_response_size` setting (64 MiB by default) limits the response +body for `read content` and `read response`, both as received and after text +decoding. The limit includes chunked responses with no declared length. Outside +a `main loop`, the connection and body read share the script's remaining +`timeout_seconds`; inside a lifetime-exempt `main loop`, each request gets a +fresh timeout of that duration. Cooperative cancellation also interrupts a +request that is waiting on the remote peer. + **Note:** inside an `open url` statement the words `method`, `headers`, and `body` introduce clauses, so use different variable names there (e.g. `request_headers`, `payload`). diff --git a/Docs/reference/configuration-reference.md b/Docs/reference/configuration-reference.md index 0798d33b..e9938070 100644 --- a/Docs/reference/configuration-reference.md +++ b/Docs/reference/configuration-reference.md @@ -211,7 +211,7 @@ All keys currently loaded from config files, with defaults. | `web_server_tls_cert_file` | path | *(none)* | Default PEM cert for bare `listen … secured` | | `web_server_tls_key_file` | path | *(none)* | Default PEM key for bare `listen … secured` | | `web_server_max_body_size` | integer ≥ 1 | `1048576` (1 MiB) | Max HTTP request body size (bytes); enforced while streaming (chunked-safe) | -| `web_server_max_response_size` | integer ≥ 1 | `67108864` (64 MiB) | Max HTTP response body size (bytes) | +| `web_server_max_response_size` | integer ≥ 1 | `67108864` (64 MiB) | Max handler or outbound HTTP response body size (bytes) | | `web_server_request_queue_bound` | integer ≥ 1 | `256` | Max queued HTTP requests before shedding with 503 | | `web_server_response_timeout_seconds` | integer ≥ 0 | `300` | Seconds to await a handler before shedding with 504; `0` disables | | `web_socket_queue_bound` | integer ≥ 1 | `1024` | Max queued frames/events per WebSocket channel before shedding | @@ -248,7 +248,11 @@ budget. #### `timeout_seconds` -Maximum execution time for a WFL script in seconds. The script terminates if it exceeds this limit. +Maximum execution time for a WFL script in seconds. Outside a `main loop`, an +outbound `open url` request (connection, headers, and response body) consumes +the run's remaining time. A `main loop` remains exempt from the lifetime limit, +but each outbound request inside it gets this duration as a fresh finite timeout +so a stalled remote peer cannot wedge the server indefinitely. - **Type:** Integer (minimum: 1) - **Default:** `60` @@ -524,7 +528,12 @@ Because request handlers run one at a time (see [Web Servers → Limitations](.. #### `web_server_max_response_size` -Maximum HTTP response body a handler may `respond with`, in bytes. A larger response is refused (the handler gets a runtime error) rather than streaming an unbounded payload to the client. +Maximum HTTP response body size, in bytes, for both directions: content a +handler may `respond with`, and content an outbound `open url` statement may +read. A larger handler response is refused; a larger outbound response is +stopped when either its received bytes or decoded UTF-8 text reaches this +limit. This applies even when the remote server uses chunked transfer encoding +or omits `Content-Length`. - **Type:** Integer (bytes, at least 1) - **Default:** `67108864` (64 MiB) diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index e6a8948d..dff389f3 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -1881,12 +1881,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls 0.26.4", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] @@ -3145,6 +3147,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" version = "0.3.103" @@ -3192,6 +3207,7 @@ dependencies = [ "bytes", "chrono", "codespan-reporting", + "encoding_rs", "futures-util", "glob", "hkdf 0.12.4", diff --git a/src/config.rs b/src/config.rs index 66340afa..e84ee434 100644 --- a/src/config.rs +++ b/src/config.rs @@ -53,9 +53,10 @@ pub struct WflConfig { /// server sheds new requests with a 503 instead of growing memory without /// bound. Default 256; must be at least 1. pub web_server_request_queue_bound: usize, - /// Maximum HTTP response body size in bytes. A handler that tries to send a - /// larger body is refused with a 500 rather than streaming an unbounded - /// payload. Feeds `ExecutionBudget`. Default 64 MiB. + /// Maximum HTTP response body size in bytes, for both handler responses and + /// bodies read by outbound `open url` statements. A larger body is refused + /// rather than buffered/streamed without bound. Feeds `ExecutionBudget`. + /// Default 64 MiB. pub web_server_max_response_size: usize, /// Maximum seconds the transport waits for a handler to answer an accepted /// HTTP request before shedding it with 504 and releasing its in-flight diff --git a/src/exec/budget.rs b/src/exec/budget.rs index 0498768c..27c8c6de 100644 --- a/src/exec/budget.rs +++ b/src/exec/budget.rs @@ -100,7 +100,8 @@ pub struct BudgetLimits { /// Maximum accepted HTTP request body size in bytes. Mapped from `.wflcfg` /// `web_server_max_body_size`. pub max_request_body_bytes: usize, - /// Maximum HTTP response body size in bytes. Mapped from `.wflcfg` + /// Maximum HTTP response body size in bytes, for both handler responses and + /// bodies read by outbound `open url` statements. Mapped from `.wflcfg` /// `web_server_max_response_size`. pub max_response_bytes: usize, /// Maximum accepted-but-unhandled HTTP requests held in the transport diff --git a/src/interpreter/mod.rs b/src/interpreter/mod.rs index 8eb53a69..96416d43 100644 --- a/src/interpreter/mod.rs +++ b/src/interpreter/mod.rs @@ -1145,6 +1145,47 @@ pub struct IoClient { config: Arc, } +/// Errors raised while an outbound HTTP request is in flight. +/// +/// Budget failures stay structured until the interpreter can attach source +/// location and the appropriate [`ErrorKind`]. Keeping them out of strings is +/// important for `try`/`when` handlers that distinguish timeouts from resource +/// limits. +#[derive(Debug)] +enum HttpClientError { + Request(String), + Budget(BudgetExceeded), + Timeout { seconds: u64 }, +} + +impl From for HttpClientError { + fn from(exceeded: BudgetExceeded) -> Self { + Self::Budget(exceeded) + } +} + +/// Which finite wall-clock limit applies to an outbound request. +#[derive(Debug, Clone, Copy)] +enum OutboundHttpDeadline { + /// An explicitly-unlimited non-server run has no wall-clock deadline. + None, + /// Outside a `main loop`, an outbound request consumes the run's remaining + /// global execution time. + Execution { + remaining: Duration, + limit_secs: u64, + }, + /// A `main loop` is lifetime-exempt, but each individual outbound request + /// still gets a fresh finite timeout so a stalled peer cannot wedge the + /// server forever. + MainLoop { duration: Duration }, +} + +/// Polling is used because cooperative cancellation is represented by an +/// atomic flag rather than a notification primitive. This interval bounds how +/// quickly an in-flight socket operation observes `ExecutionBudget::cancel()`. +const HTTP_CANCELLATION_POLL_INTERVAL: Duration = Duration::from_millis(10); + impl IoClient { fn new(config: Arc) -> Self { Self { @@ -1197,31 +1238,32 @@ impl IoClient { } #[allow(dead_code)] - async fn http_get(&self, url: &str) -> Result { - match self.http_client.get(url).send().await { - Ok(response) => match response.text().await { - Ok(text) => Ok(text), - Err(e) => Err(format!("Failed to read response body: {e}")), - }, - Err(e) => Err(format!("Failed to send HTTP GET request: {e}")), - } + async fn http_get( + &self, + url: &str, + budget: Arc, + ) -> Result { + let (_, _, body) = self + .send_http_request(self.http_client.get(url), "GET", budget) + .await?; + Ok(body) } #[allow(dead_code)] - async fn http_post(&self, url: &str, data: &str) -> Result { - match self - .http_client - .post(url) - .body(data.to_string()) - .send() - .await - { - Ok(response) => match response.text().await { - Ok(text) => Ok(text), - Err(e) => Err(format!("Failed to read response body: {e}")), - }, - Err(e) => Err(format!("Failed to send HTTP POST request: {e}")), - } + async fn http_post( + &self, + url: &str, + data: &str, + budget: Arc, + ) -> Result { + let (_, _, body) = self + .send_http_request( + self.http_client.post(url).body(data.to_string()), + "POST", + budget, + ) + .await?; + Ok(body) } /// Perform an HTTP request with an arbitrary method, optional headers, @@ -1233,9 +1275,10 @@ impl IoClient { url: &str, headers: &[(String, String)], body: Option, - ) -> Result<(u16, Vec<(String, String)>, String), String> { + budget: Arc, + ) -> Result<(u16, Vec<(String, String)>, String), HttpClientError> { let parsed_method = reqwest::Method::from_bytes(method.as_bytes()) - .map_err(|_| format!("Invalid HTTP method: {method}"))?; + .map_err(|_| HttpClientError::Request(format!("Invalid HTTP method: {method}")))?; let mut request = self.http_client.request(parsed_method, url); for (name, value) in headers { @@ -1245,28 +1288,244 @@ impl IoClient { request = request.body(body); } - match request.send().await { - Ok(response) => { - let status = response.status().as_u16(); - // Header names are normalized to lowercase for consistent - // access from WFL (e.g. resp.headers["content-type"]), and - // non-UTF8 values are converted lossily instead of dropped. - let response_headers = response - .headers() - .iter() - .map(|(name, value)| { - ( - name.as_str().to_ascii_lowercase(), - String::from_utf8_lossy(value.as_bytes()).into_owned(), - ) - }) - .collect(); - match response.text().await { - Ok(text) => Ok((status, response_headers, text)), - Err(e) => Err(format!("Failed to read response body: {e}")), + self.send_http_request(request, method, budget).await + } + + /// Send a request and consume its body without ever buffering more than the + /// configured response ceiling. The budget passed here is deliberately the + /// interpreter's *live* budget, not construction-time IoClient state: the + /// REPL replaces its budget for every command. + async fn send_http_request( + &self, + request: reqwest::RequestBuilder, + method: &str, + budget: Arc, + ) -> Result<(u16, Vec<(String, String)>, String), HttpClientError> { + let method = method.to_string(); + let operation_budget = Arc::clone(&budget); + let operation = async move { + use futures_util::StreamExt; + + let response = request.send().await.map_err(|e| { + HttpClientError::Request(format!("Failed to send HTTP {method} request: {e}")) + })?; + + let status = response.status().as_u16(); + // Header names are normalized to lowercase for consistent access + // from WFL (e.g. resp.headers["content-type"]), and non-UTF8 + // values are converted lossily instead of dropped. + let response_headers = response + .headers() + .iter() + .map(|(name, value)| { + ( + name.as_str().to_ascii_lowercase(), + String::from_utf8_lossy(value.as_bytes()).into_owned(), + ) + }) + .collect(); + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + + let max_response_bytes = operation_budget.limits().max_response_bytes; + if let Some(content_length) = response.content_length() { + let max_as_u64 = u64::try_from(max_response_bytes).unwrap_or(u64::MAX); + if content_length > max_as_u64 { + return Err(HttpClientError::Budget(BudgetExceeded::ResponseBytes { + limit: max_response_bytes, + actual: usize::try_from(content_length).unwrap_or(usize::MAX), + })); + } + } + + // Do not retain a full raw byte buffer and then allocate a second, + // potentially larger UTF-8 string. Decode each network chunk into + // bounded scratch space, and enforce the same ceiling on both wire + // bytes and decoded UTF-8 bytes. Invalid UTF-8 alone can expand 3x + // when replaced with U+FFFD. + let initial_capacity = response + .content_length() + .and_then(|len| usize::try_from(len).ok()) + .unwrap_or(0) + .min(max_response_bytes) + .min(64 * 1024); + let encoding = Self::http_text_encoding(content_type.as_deref()); + let mut decoder = encoding.new_decoder(); + let mut body = String::with_capacity(initial_capacity); + let mut wire_bytes = 0_usize; + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| { + HttpClientError::Request(format!("Failed to read response body: {e}")) + })?; + let actual = wire_bytes.saturating_add(chunk.len()); + if chunk.len() > max_response_bytes.saturating_sub(wire_bytes) { + return Err(HttpClientError::Budget(BudgetExceeded::ResponseBytes { + limit: max_response_bytes, + actual, + })); + } + wire_bytes = actual; + Self::decode_http_chunk( + &mut decoder, + &chunk, + false, + &mut body, + max_response_bytes, + )?; + } + Self::decode_http_chunk(&mut decoder, &[], true, &mut body, max_response_bytes)?; + + Ok((status, response_headers, body)) + }; + + // A custom live budget may deliberately have no run-wide deadline. In + // a lifetime-exempt main loop, fall back to the interpreter's + // configured `timeout_seconds` (minimum one second) so the individual + // network operation is still finite and user-configurable. + let configured_timeout = Duration::from_secs(self.config.timeout_seconds.max(1)); + Self::run_http_with_budget(budget, configured_timeout, operation).await + } + + /// Select the response encoding while preserving reqwest's text behavior: + /// honor a declared charset and default to UTF-8. + fn http_text_encoding(content_type: Option<&str>) -> &'static encoding_rs::Encoding { + let charset = content_type.and_then(|value| { + value.split(';').skip(1).find_map(|parameter| { + let (name, value) = parameter.trim().split_once('=')?; + name.trim() + .eq_ignore_ascii_case("charset") + .then(|| value.trim().trim_matches(|ch| ch == '\'' || ch == '"')) + }) + }); + charset + .and_then(|name| encoding_rs::Encoding::for_label(name.as_bytes())) + .unwrap_or(encoding_rs::UTF_8) + } + + /// Incrementally decode one response chunk into caller-owned text. The + /// decoder writes only into fixed scratch storage; the destination reserves + /// exactly the accepted addition before appending, avoiding Vec/String + /// geometric-growth spikes near the configured ceiling. + fn decode_http_chunk( + decoder: &mut encoding_rs::Decoder, + mut input: &[u8], + last: bool, + output: &mut String, + max_response_bytes: usize, + ) -> Result<(), HttpClientError> { + let mut decoded = [0_u8; 8 * 1024]; + loop { + let (result, read, written, _) = decoder.decode_to_utf8(input, &mut decoded, last); + let actual = output.len().saturating_add(written); + if written > max_response_bytes.saturating_sub(output.len()) { + return Err(HttpClientError::Budget(BudgetExceeded::ResponseBytes { + limit: max_response_bytes, + actual, + })); + } + if written > 0 { + output.try_reserve_exact(written).map_err(|error| { + HttpClientError::Request(format!( + "Failed to allocate bounded HTTP response buffer: {error}" + )) + })?; + let text = std::str::from_utf8(&decoded[..written]) + .expect("encoding_rs must emit valid UTF-8"); + output.push_str(text); + } + input = &input[read..]; + + match result { + encoding_rs::CoderResult::InputEmpty => return Ok(()), + encoding_rs::CoderResult::OutputFull => {} + } + } + } + + fn outbound_http_deadline( + budget: &ExecutionBudget, + configured_timeout: Duration, + ) -> Result { + budget.check_cancelled()?; + + if budget.is_deadline_exempt() { + return Ok(OutboundHttpDeadline::MainLoop { + duration: budget.limits().max_duration.unwrap_or(configured_timeout), + }); + } + + let Some(limit) = budget.limits().max_duration else { + return Ok(OutboundHttpDeadline::None); + }; + let Some(remaining) = limit.checked_sub(budget.elapsed()) else { + return Err(HttpClientError::Budget(BudgetExceeded::Deadline { + limit_secs: limit.as_secs(), + })); + }; + if remaining.is_zero() { + return Err(HttpClientError::Budget(BudgetExceeded::Deadline { + limit_secs: limit.as_secs(), + })); + } + Ok(OutboundHttpDeadline::Execution { + remaining, + limit_secs: limit.as_secs(), + }) + } + + /// Race the complete network operation (connect, headers, and streamed + /// body) against both cooperative cancellation and the applicable finite + /// deadline. Dropping reqwest's future closes/cancels the in-flight work. + async fn run_http_with_budget( + budget: Arc, + configured_timeout: Duration, + operation: F, + ) -> Result + where + F: std::future::Future>, + { + let deadline = Self::outbound_http_deadline(&budget, configured_timeout)?; + let timeout_duration = match deadline { + OutboundHttpDeadline::None => None, + OutboundHttpDeadline::Execution { remaining, .. } => Some(remaining), + OutboundHttpDeadline::MainLoop { duration } => Some(duration), + }; + + let cancellation_budget = Arc::clone(&budget); + let cancellation = async move { + loop { + if cancellation_budget.is_cancelled() { + break; } + tokio::time::sleep(HTTP_CANCELLATION_POLL_INTERVAL).await; + } + }; + let timeout = async move { + match timeout_duration { + Some(duration) => tokio::time::sleep(duration).await, + None => std::future::pending::<()>().await, } - Err(e) => Err(format!("Failed to send HTTP {method} request: {e}")), + }; + + tokio::pin!(operation); + tokio::pin!(cancellation); + tokio::pin!(timeout); + tokio::select! { + result = &mut operation => result, + _ = &mut cancellation => Err(HttpClientError::Budget(BudgetExceeded::Cancelled)), + _ = &mut timeout => match deadline { + OutboundHttpDeadline::Execution { limit_secs, .. } => { + Err(HttpClientError::Budget(BudgetExceeded::Deadline { limit_secs })) + } + OutboundHttpDeadline::MainLoop { duration } => { + Err(HttpClientError::Timeout { seconds: duration.as_secs() }) + } + OutboundHttpDeadline::None => unreachable!("disabled timeout cannot complete"), + }, } } @@ -2614,6 +2873,26 @@ impl Interpreter { RuntimeError::with_kind(exceeded.message(), line, column, kind) } + /// Attach WFL source information to an outbound HTTP error while + /// preserving structured timeout/resource-limit kinds for `try`/`when`. + fn http_client_error( + &self, + error: HttpClientError, + line: usize, + column: usize, + ) -> RuntimeError { + match error { + HttpClientError::Request(message) => RuntimeError::new(message, line, column), + HttpClientError::Budget(exceeded) => self.budget_error(exceeded, line, column), + HttpClientError::Timeout { seconds } => RuntimeError::with_kind( + format!("Outbound HTTP request exceeded timeout ({seconds}s)"), + line, + column, + ErrorKind::Timeout, + ), + } + } + /// Map a pattern-VM error onto a `RuntimeError`. Budget breaches (step/state /// ceilings, cancellation) surface as catchable `ResourceLimit` errors so a /// ReDoS/cancellation during matching is not silently collapsed into a @@ -5008,7 +5287,11 @@ impl Interpreter { } }; - match self.io_client.http_get(&url_str).await { + match self + .io_client + .http_get(&url_str, Arc::clone(&self.budget)) + .await + { Ok(body) => { match env .borrow_mut() @@ -5018,7 +5301,7 @@ impl Interpreter { Err(msg) => Err(RuntimeError::new(msg, *line, *column)), } } - Err(e) => Err(RuntimeError::new(e, *line, *column)), + Err(error) => Err(self.http_client_error(error, *line, *column)), } } Statement::HttpPostStatement { @@ -5053,7 +5336,11 @@ impl Interpreter { } }; - match self.io_client.http_post(&url_str, &data_str).await { + match self + .io_client + .http_post(&url_str, &data_str, Arc::clone(&self.budget)) + .await + { Ok(body) => { match env .borrow_mut() @@ -5063,7 +5350,7 @@ impl Interpreter { Err(msg) => Err(RuntimeError::new(msg, *line, *column)), } } - Err(e) => Err(RuntimeError::new(e, *line, *column)), + Err(error) => Err(self.http_client_error(error, *line, *column)), } } Statement::HttpRequestStatement { @@ -5171,7 +5458,13 @@ impl Interpreter { match self .io_client - .http_request(&method_str, &url_str, &header_list, body_str) + .http_request( + &method_str, + &url_str, + &header_list, + body_str, + Arc::clone(&self.budget), + ) .await { Ok((status, response_headers, response_body)) => { @@ -5203,7 +5496,7 @@ impl Interpreter { Err(msg) => Err(RuntimeError::new(msg, *line, *column)), } } - Err(e) => Err(RuntimeError::new(e, *line, *column)), + Err(error) => Err(self.http_client_error(error, *line, *column)), } } Statement::RepeatWhileLoop { diff --git a/tests/http_outbound_budget_test.rs b/tests/http_outbound_budget_test.rs new file mode 100644 index 00000000..e8a1a6f5 --- /dev/null +++ b/tests/http_outbound_budget_test.rs @@ -0,0 +1,298 @@ +//! Security regressions for bounded outbound HTTP responses. +//! +//! These tests use a minimal local TCP peer so they are deterministic and do +//! not require internet access. Together they exercise all three runtime paths: +//! legacy GET, legacy POST, and the arbitrary-method/full-response statement. + +use std::sync::Arc; +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::oneshot; + +use wfl::config::WflConfig; +use wfl::exec::budget::{BudgetLimits, ExecutionBudget}; +use wfl::interpreter::Interpreter; +use wfl::interpreter::error::ErrorKind; +use wfl::lexer::lex_wfl_with_positions; +use wfl::parser::Parser; +use wfl::parser::ast::{Expression, Literal, Program, Statement}; + +fn parse(source: &str) -> Program { + let tokens = lex_wfl_with_positions(source); + Parser::new(&tokens) + .parse() + .unwrap_or_else(|errors| panic!("WFL source should parse: {errors:?}")) +} + +async fn read_request_headers(socket: &mut TcpStream) { + let mut request = Vec::new(); + let mut chunk = [0_u8; 1024]; + loop { + let read = socket + .read(&mut chunk) + .await + .expect("read local HTTP request"); + assert!(read > 0, "client closed before completing HTTP headers"); + request.extend_from_slice(&chunk[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + return; + } + assert!( + request.len() <= 64 * 1024, + "unexpectedly large test request" + ); + } +} + +/// Spawn a one-shot HTTP peer that writes `response_prefix`. When `stall` is +/// true it keeps the connection open afterward instead of completing the body. +async fn spawn_http_peer( + response_prefix: &'static [u8], + stall: bool, +) -> ( + String, + tokio::task::JoinHandle<()>, + oneshot::Receiver>, +) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind local HTTP peer"); + let address = listener.local_addr().expect("local HTTP peer address"); + let (response_attempted, response_attempted_rx) = oneshot::channel(); + let handle = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept HTTP request"); + read_request_headers(&mut socket).await; + let write_result = socket + .write_all(response_prefix) + .await + .map_err(|error| error.kind()); + let _ = response_attempted.send(write_result); + if stall { + std::future::pending::<()>().await; + } + let _ = socket.shutdown().await; + }); + (format!("http://{address}"), handle, response_attempted_rx) +} + +async fn await_http_peer(server: tokio::task::JoinHandle<()>) { + tokio::time::timeout(Duration::from_secs(1), server) + .await + .expect("local HTTP peer must not hang") + .expect("local HTTP peer task must not panic"); +} + +fn assert_response_limit(errors: &[wfl::interpreter::error::RuntimeError], limit: usize) { + let error = errors.first().expect("one runtime error"); + assert_eq!(error.kind, ErrorKind::ResourceLimit); + assert!( + error.message.contains("Response body too large"), + "expected response-size diagnostic, got: {error:?}" + ); + assert!( + error.message.contains(&format!("limit: {limit} bytes")), + "diagnostic should include the configured limit: {error:?}" + ); +} + +#[tokio::test] +async fn legacy_get_rejects_oversized_content_length() { + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: text/plain\r\n\ +Content-Length: 32\r\n\ +Connection: close\r\n\ +\r\n\ +0123456789abcdef0123456789abcdef"; + let (url, server, _response_attempted) = spawn_http_peer(response, false).await; + + let config = WflConfig { + web_server_max_response_size: 8, + ..Default::default() + }; + let mut interpreter = Interpreter::with_config(Arc::new(config)); + let program = parse(&format!( + r#"open url at "{url}" and read content as content"# + )); + + let errors = interpreter + .interpret(&program) + .await + .expect_err("advertised response above the cap must fail"); + assert_response_limit(&errors, 8); + await_http_peer(server).await; +} + +#[tokio::test] +async fn full_response_request_rejects_oversized_chunked_body() { + // The response has no Content-Length, so only incremental accounting can + // catch that the decoded body grows from five to nine bytes over an 8-byte + // cap. + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: text/plain\r\n\ +Transfer-Encoding: chunked\r\n\ +Connection: close\r\n\ +\r\n\ +5\r\n12345\r\n\ +4\r\n6789\r\n\ +0\r\n\r\n"; + let (url, server, _response_attempted) = spawn_http_peer(response, false).await; + + let config = WflConfig { + web_server_max_response_size: 8, + ..Default::default() + }; + let mut interpreter = Interpreter::with_config(Arc::new(config)); + let program = parse(&format!( + r#"open url at "{url}" and read response as reply"# + )); + + let errors = interpreter + .interpret(&program) + .await + .expect_err("chunked response above the cap must fail"); + assert_response_limit(&errors, 8); + await_http_peer(server).await; +} + +#[tokio::test] +async fn decoded_text_cannot_expand_past_the_response_limit() { + // Four malformed UTF-8 bytes decode to four three-byte replacement + // characters. The wire body fits the four-byte limit; the decoded text + // must still be rejected before it can expand beyond that same ceiling. + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: text/plain; charset=utf-8\r\n\ +Content-Length: 4\r\n\ +Connection: close\r\n\ +\r\n\ +\xff\xff\xff\xff"; + let (url, server, _response_attempted) = spawn_http_peer(response, false).await; + + let config = WflConfig { + web_server_max_response_size: 4, + ..Default::default() + }; + let mut interpreter = Interpreter::with_config(Arc::new(config)); + let program = parse(&format!( + r#"open url at "{url}" and read content as content"# + )); + + let errors = interpreter + .interpret(&program) + .await + .expect_err("decoded response expansion above the cap must fail"); + assert_response_limit(&errors, 4); + await_http_peer(server).await; +} + +#[tokio::test] +async fn legacy_post_body_read_observes_cooperative_cancellation() { + // Headers promise five bytes, but the peer never sends them. Before the + // fix, Response::text() could remain parked here indefinitely. + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: text/plain\r\n\ +Content-Length: 5\r\n\ +Connection: close\r\n\ +\r\n"; + let (url, server, response_attempted) = spawn_http_peer(response, true).await; + + let mut interpreter = Interpreter::new(); + let budget = interpreter.budget(); + // Construct the legacy node directly: `with data` is no longer accepted by + // the current grammar, but embedded/previously parsed programs still reach + // the dedicated HttpPostStatement execution path. + let program = Program { + statements: vec![Statement::HttpPostStatement { + url: Expression::Literal(Literal::String(url.into()), 1, 1), + data: Expression::Literal(Literal::String("x=1".into()), 1, 1), + variable_name: "reply".to_string(), + line: 1, + column: 1, + }], + }; + let interpret = interpreter.interpret(&program); + tokio::pin!(interpret); + tokio::time::timeout(Duration::from_secs(1), async { + tokio::select! { + result = response_attempted => { + assert_eq!( + result.expect("local HTTP peer reports its response write"), + Ok(()), + "the cancellation regression must reach the stalled body read" + ); + budget.cancel(); + } + result = &mut interpret => { + panic!("request completed before the peer entered its stalled body: {result:?}"); + } + } + }) + .await + .expect("request must reach the peer's stalled response promptly"); + + let errors = tokio::time::timeout(Duration::from_secs(1), &mut interpret) + .await + .expect("in-flight POST should observe cancellation promptly") + .expect_err("cancelled POST must fail"); + let error = errors.first().expect("one runtime error"); + assert_eq!(error.kind, ErrorKind::ResourceLimit); + assert!( + error.message.contains("Execution was cancelled"), + "expected cancellation diagnostic, got: {error:?}" + ); + + server.abort(); + let _ = server.await; +} + +#[tokio::test] +async fn main_loop_gives_each_outbound_request_a_finite_timeout() { + // A main loop is deliberately exempt from the run-lifetime deadline. An + // individual outbound request inside it must still reuse that configured + // duration as a fresh per-request limit. + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: text/plain\r\n\ +Content-Length: 5\r\n\ +Connection: close\r\n\ +\r\n"; + let (url, server, _response_attempted) = spawn_http_peer(response, true).await; + + let config = Arc::new(WflConfig::default()); + let mut interpreter = Interpreter::with_config(Arc::clone(&config)); + let program = parse(&format!( + r#" +main loop: + open url at "{url}" and read content as content + break +end loop +"# + )); + + // Install the deliberately short budget only after building the HTTP + // client and parsing the fixture. The budget's start instant covers the + // whole run, so including unrelated setup here can exhaust it before the + // interpreter enters the main loop on slower CI hosts. This test is about + // the fresh per-request deadline applied *inside* that loop. + let limits = BudgetLimits { + max_duration: Some(Duration::from_millis(250)), + ..BudgetLimits::from_config(&config) + }; + interpreter.set_budget(Arc::new(ExecutionBudget::new(limits))); + + let errors = tokio::time::timeout(Duration::from_secs(2), interpreter.interpret(&program)) + .await + .expect("main-loop request should have a finite timeout") + .expect_err("stalled main-loop request must time out"); + let error = errors.first().expect("one runtime error"); + assert_eq!(error.kind, ErrorKind::Timeout); + assert!( + error + .message + .contains("Outbound HTTP request exceeded timeout"), + "expected outbound timeout diagnostic, got: {error:?}" + ); + + server.abort(); + let _ = server.await; +}