From ec505db147e9f11e98de11511a388e67ee63aeb3 Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Tue, 8 Sep 2026 18:14:42 -0700 Subject: [PATCH] fix(middleware): address HTTP response review findings Signed-off-by: Piotr Mlocek --- architecture/sandbox.md | 7 +- .../src/lib.rs | 3 +- .../src/response.rs | 370 ++++++++++++++- .../src/l7/rest.rs | 443 ++++++++++++++++-- docs/reference/gateway-config.mdx | 2 + .../README.md | 12 +- .../smoke.sh | 26 +- skills/generate-sandbox-policy/SKILL.md | 2 +- tasks/scripts/stage-prebuilt-binaries.sh | 4 +- 9 files changed, 794 insertions(+), 75 deletions(-) diff --git a/architecture/sandbox.md b/architecture/sandbox.md index 81c4a1b76..db443bc79 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -199,7 +199,12 @@ can change. Whole-body stages delay commitment and share one non-resetting, supervisor-wide accumulation deadline. Body stages receive a final body result and then one trailer exchange; trailer mutations can only change or remove existing, non-protected names. Intentional blocks return the canonical 403 -before commitment and abort delivery without injected bytes after commitment. +before commitment and abort delivery without injected bytes after commitment. Streaming input units flush after bounded coalescing even within a +content-length body or transfer chunk. Coalescing cancels only input acquisition; +deadline transitions and downstream writes finish outside those timeouts. +The response runtime caps aggregate retained body data across stages and pending +output at 8 MiB. A transformation that exceeds the budget follows its stage's +failure policy, preserving its input when failing open. The supervisor installs policy and middleware registry changes as one runtime generation and preserves the last-known-good generation if preparation fails. diff --git a/crates/openshell-supervisor-middleware/src/lib.rs b/crates/openshell-supervisor-middleware/src/lib.rs index 57d350cd4..dcf695539 100644 --- a/crates/openshell-supervisor-middleware/src/lib.rs +++ b/crates/openshell-supervisor-middleware/src/lib.rs @@ -11,7 +11,8 @@ mod websocket; pub use response::{ HttpResponseDiagnostics, HttpResponseFinish, HttpResponseInvocation, HttpResponseInvocationOutcome, HttpResponseMiddlewareFailure, HttpResponsePreflightInput, - HttpResponsePreflightOutcome, HttpResponseSession, MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES, + HttpResponsePreflightOutcome, HttpResponseSession, MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES, + MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES, }; pub use websocket::{ diff --git a/crates/openshell-supervisor-middleware/src/response.rs b/crates/openshell-supervisor-middleware/src/response.rs index 2504b1a14..225df45e8 100644 --- a/crates/openshell-supervisor-middleware/src/response.rs +++ b/crates/openshell-supervisor-middleware/src/response.rs @@ -32,6 +32,9 @@ use super::{ const STREAM_CHANNEL_CAPACITY: usize = 4; pub const MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES: usize = 64 * 1024; +/// Maximum logical body bytes retained across a session's stage buffers and +/// pending output. Temporary exchange copies have the per-binding payload cap. +pub const MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES: usize = 8 * 1024 * 1024; #[derive(Debug, Clone)] pub struct HttpResponsePreflightInput { @@ -179,6 +182,7 @@ pub struct HttpResponseSession { invocations: Vec, session_admission: Option, body_transformed: bool, + retained_body_bytes: usize, defer_output_until_finish: bool, deferred_output: Vec>, connection_nominated_headers: Vec, @@ -242,6 +246,7 @@ impl HttpResponseSession { } } self.defer_output_until_finish = false; + self.release_body_bytes(&released); Ok(released) } @@ -285,8 +290,16 @@ impl HttpResponseSession { diagnostics: HttpResponseDiagnostics::default(), })?; let deadline = Instant::now() + MAX_MIDDLEWARE_CHAIN_TIMEOUT; + // Between pushes, the first active whole-body barrier owns all input + // not returned to the relay (at most the 4 MiB binding cap). Later + // barriers cannot receive bytes until it finishes or disables itself; + // finish consumes the session and expiry disables all such barriers. + // Replacement admission reserves an additional upstream unit below. + self.retained_body_bytes += data.len(); + debug_assert!(self.retained_body_bytes <= MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES); let output = self.process_units_from(0, vec![data], deadline).await?; if !self.defer_output_until_finish { + self.release_body_bytes(&output); return Ok(output); } if self.requires_whole_body() { @@ -297,6 +310,7 @@ impl HttpResponseSession { self.defer_output_until_finish = false; let mut released = std::mem::take(&mut self.deferred_output); released.extend(output); + self.release_body_bytes(&released); Ok(released) } @@ -369,6 +383,10 @@ impl HttpResponseSession { self.end_all(reason).await; } + fn release_body_bytes(&mut self, units: &[Vec]) { + self.retained_body_bytes -= units.iter().map(Vec::len).sum::(); + } + async fn process_units_from( &mut self, start: usize, @@ -526,6 +544,31 @@ impl HttpResponseSession { } }; let input_size = original.len(); + let replacement_size = match &decision.action { + BodyAction::Transform(replacement) + | BodyAction::SkipRemaining(CurrentBodyAction::Transform(replacement)) => { + Some(replacement.len()) + } + _ => None, + }; + if let Some(replacement_size) = replacement_size { + let retained = self.retained_body_bytes - input_size + replacement_size; + // Reserve room for one more normalized upstream unit. Whole-body + // barriers bound the input retained between calls to push_body. + if retained + > MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES - MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES + { + return self + .handle_stage_failure( + index, + "response_body_aggregate_over_capacity", + Some(sequence), + original, + ) + .await; + } + self.retained_body_bytes = retained; + } let stage = &mut self.stages[index]; collect_diagnostics( stage, @@ -797,11 +840,11 @@ impl ChainRunner { entries: &[ChainEntry], input: HttpResponsePreflightInput, ) -> miette::Result { - validate_preflight_input(&input)?; let described = self.describe_http_response_chain(entries).await?; if described.is_empty() { return Ok(empty_preflight_outcome(input.headers)); } + validate_preflight_input(&input)?; let session_admission = match self.try_reserve_middleware_session() { MiddlewareSessionAdmission::Admitted(admission) => admission, MiddlewareSessionAdmission::AtCapacity => { @@ -852,16 +895,16 @@ impl ChainRunner { }; let timeout = entry.timeout; let opened = tokio::time::timeout(timeout, async { - let mut responses = service - .service - .open_http_response_pre_return(receiver) - .await?; sender .send(HttpResponseEvent { event: Some(http_response_event::Event::Preflight(preflight)), }) .await .map_err(|_| tonic::Status::unavailable("middleware request stream closed"))?; + let mut responses = service + .service + .open_http_response_pre_return(receiver) + .await?; let response = responses.next().await.ok_or_else(|| { tonic::Status::unavailable("middleware result stream closed") })??; @@ -1163,6 +1206,7 @@ impl ChainRunner { invocations: Vec::new(), session_admission: Some(session_admission), body_transformed: false, + retained_body_bytes: 0, defer_output_until_finish, deferred_output: Vec::new(), connection_nominated_headers: input.connection_nominated_headers, @@ -1811,6 +1855,9 @@ mod tests { Configured, HangBody, LargeStream, + Expansion, + DeleteBody, + SkipBody, Skip, InvalidSkipReason, TrailerMutation, @@ -1883,6 +1930,13 @@ mod tests { request: tonic::Request>, ) -> Result, tonic::Status> { let mut requests = request.into_inner(); + // Exercise servers that inspect the initial request before sending + // response headers, rather than returning a stream immediately. + let first = requests.next().await.expect("initial request"); + assert!(matches!(&first, Ok(HttpResponseEvent { + event: Some(http_response_event::Event::Preflight(_)) + }))); + let mut requests = futures::stream::iter([first]).chain(requests); let (sender, receiver) = mpsc::channel(4); tokio::spawn(async move { while let Some(Ok(event)) = requests.next().await { @@ -1932,7 +1986,10 @@ mod tests { operation: openshell_core::proto::SupervisorMiddlewareOperation::HttpResponse as i32, phase: openshell_core::proto::SupervisorMiddlewarePhase::PreReturn as i32, - max_payload_bytes: if matches!(self.script, Script::LargeStream) { + max_payload_bytes: if matches!( + self.script, + Script::LargeStream | Script::Expansion + ) { 128 * 1024 } else { 4096 @@ -2033,6 +2090,9 @@ mod tests { | Script::InvalidSequence | Script::HangBody | Script::LargeStream + | Script::Expansion + | Script::DeleteBody + | Script::SkipBody | Script::TrailerMutation | Script::InvalidTrailerMutation => { (HttpResponseBodyMode::StreamBytes, Vec::new()) @@ -2072,6 +2132,9 @@ mod tests { break; }; let replacement = match selected_script { + Script::Expansion => vec![b'x'; 128 * 1024], + Script::DeleteBody => Vec::new(), + Script::SkipBody => b"replacement".to_vec(), Script::Stream | Script::InvalidSequence | Script::LargeStream @@ -2084,6 +2147,24 @@ mod tests { | Script::Skip | Script::InvalidSkipReason => break, }; + let transform = HttpResponseBodyTransform { + replacement: Some(http_response_body_transform::Replacement::Data( + replacement, + )), + }; + let action = if matches!(selected_script, Script::SkipBody) { + http_response_body_result::Action::SkipRemaining( + openshell_core::proto::HttpResponseBodySkipRemaining { + current: Some( + http_response_body_skip_remaining::Current::Transform( + transform, + ), + ), + }, + ) + } else { + http_response_body_result::Action::Transform(transform) + }; HttpResponseEventResult { result: Some(http_response_event_result::Result::BodyResult( HttpResponseBodyResult { @@ -2095,15 +2176,7 @@ mod tests { } else { body.sequence }, - action: Some(http_response_body_result::Action::Transform( - HttpResponseBodyTransform { - replacement: Some( - http_response_body_transform::Replacement::Data( - replacement, - ), - ), - }, - )), + action: Some(action), ..Default::default() }, )), @@ -2233,6 +2306,270 @@ mod tests { assert!(!modes.contains(&(HttpResponseBodyMode::StreamBytes as i32))); } + struct ReadPreflightBeforeOpening { + failure: Option, + } + + #[tonic::async_trait] + impl InProcessMiddleware for ReadPreflightBeforeOpening { + async fn describe(&self) -> MiddlewareManifest { + let mut manifest = response_manifest("test/response"); + manifest.bindings[0].timeout = "10ms".into(); + manifest + } + + async fn validate_config(&self, _: &str, _: &prost_types::Struct) -> miette::Result<()> { + Ok(()) + } + + async fn evaluate_http_request( + &self, + _: HttpRequestView<'_>, + ) -> miette::Result { + unreachable!() + } + + async fn open_http_response_pre_return( + &self, + mut requests: mpsc::Receiver, + ) -> Result { + let first = requests.recv().await.expect("initial preflight"); + assert!(matches!( + first.event, + Some(http_response_event::Event::Preflight(_)) + )); + if let Some(hang) = self.failure { + if hang { + futures::future::pending::<()>().await; + } + return Err(tonic::Status::unavailable("startup failed")); + } + let response = HttpResponseEventResult { + result: Some(http_response_event_result::Result::PreflightResult( + HttpResponsePreflightResult { + action: Some(http_response_preflight_result::Action::Inspect( + HttpResponsePreflightInspect { + body_mode: HttpResponseBodyMode::HeadersOnly as i32, + header_mutations: Vec::new(), + }, + )), + ..Default::default() + }, + )), + }; + Ok(Box::pin(futures::stream::iter([Ok(response)]))) + } + } + + #[tokio::test] + async fn preflight_can_be_read_before_open_returns() { + let runner = ChainRunner::new(Arc::new(ReadPreflightBeforeOpening { failure: None })); + let outcome = tokio::time::timeout( + Duration::from_secs(1), + runner.preflight_http_response(&[entry(OnError::FailClosed)], input(200)), + ) + .await + .expect("bounded startup") + .expect("preflight"); + assert!(outcome.allowed, "{}", outcome.reason); + } + + #[tokio::test] + async fn preflight_opening_failure_obeys_policy_and_releases_admission() { + for hang in [false, true] { + for on_error in [OnError::FailOpen, OnError::FailClosed] { + let runner = ChainRunner::new(Arc::new(ReadPreflightBeforeOpening { + failure: Some(hang), + })); + let permits = runner.registry.session_admission.available_permits(); + let outcome = tokio::time::timeout( + Duration::from_secs(1), + runner.preflight_http_response(&[entry(on_error)], input(200)), + ) + .await + .expect("bounded opening failure") + .unwrap(); + assert_eq!(outcome.allowed, on_error == OnError::FailOpen); + assert!(outcome.session.is_none()); + assert_eq!( + runner.registry.session_admission.available_permits(), + permits + ); + assert!(outcome.invocations[0].failed); + } + } + } + + #[tokio::test] + async fn multiple_whole_body_barriers_preserve_accounting_on_overflow_and_expiry() { + for expire in [false, true] { + let runner = ChainRunner::new(Arc::new(ResponseService { + script: Script::Configured, + })); + let mut entries = vec![ + configured_entry("first", 0, "whole"), + configured_entry("second", 1, "whole"), + configured_entry("stream", 2, "stream"), + ]; + for entry in &mut entries { + entry.on_error = OnError::FailOpen; + } + let mut outcome = runner + .preflight_http_response(&entries, input(200)) + .await + .unwrap(); + let mut session = outcome.session.take().unwrap(); + for _ in 0..2 { + assert!( + session + .push_body(vec![b'a'; 2048]) + .await + .unwrap() + .is_empty() + ); + assert!(session.retained_body_bytes <= 4096); + assert_eq!( + session + .stages + .iter() + .filter(|stage| !stage.whole_body.is_empty()) + .count(), + 1 + ); + } + let output = if expire { + session.start_whole_body_deadline(Duration::ZERO); + session.expire_whole_body_deadline().await.unwrap() + } else { + session.push_body(vec![b'a'; 2048]).await.unwrap() + }; + assert_eq!( + output.concat(), + vec![b'A'; if expire { 4096 } else { 6144 }] + ); + assert_eq!(session.retained_body_bytes, 0); + assert_eq!( + session.push_body(b"next".to_vec()).await.unwrap().concat(), + b"NEXT" + ); + assert_eq!(session.retained_body_bytes, 0); + assert!( + session + .finish(Vec::new()) + .await + .unwrap() + .body_units + .is_empty() + ); + } + } + + #[tokio::test] + async fn deleted_and_skip_remaining_units_release_body_accounting() { + for script in [Script::DeleteBody, Script::SkipBody] { + let runner = ChainRunner::new(Arc::new(ResponseService { script })); + let mut outcome = runner + .preflight_http_response(&[entry(OnError::FailClosed)], input(200)) + .await + .unwrap(); + let mut session = outcome.session.take().unwrap(); + for index in 0..3 { + let output = session.push_body(b"original".to_vec()).await.unwrap(); + let expected = match script { + Script::DeleteBody => Vec::new(), + Script::SkipBody if index == 0 => b"replacement".to_vec(), + Script::SkipBody => b"original".to_vec(), + _ => unreachable!(), + }; + assert_eq!(output.concat(), expected); + assert_eq!(session.retained_body_bytes, 0); + } + assert!( + session + .finish(Vec::new()) + .await + .unwrap() + .body_units + .is_empty() + ); + } + } + + #[tokio::test] + async fn expanding_stages_obey_aggregate_budget_and_failure_policy() { + for on_error in [OnError::FailClosed, OnError::FailOpen] { + let runner = ChainRunner::new(Arc::new(ResponseService { + script: Script::Expansion, + })); + let permits = runner.registry.session_admission.available_permits(); + let entries = (0..9) + .map(|order| { + let mut entry = entry(on_error); + entry.name = format!("expand-{order}"); + entry.order = order; + entry + }) + .collect::>(); + let mut outcome = runner + .preflight_http_response(&entries, input(200)) + .await + .unwrap(); + let mut session = outcome.session.take().unwrap(); + let result = tokio::time::timeout(Duration::from_secs(5), session.push_body(vec![1])) + .await + .expect("bounded expansion"); + match result { + Ok(output) => { + assert_eq!(on_error, OnError::FailOpen); + assert!( + output.iter().map(Vec::len).sum::() + < MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES + ); + assert!(output.iter().flatten().all(|byte| *byte == b'x')); + assert_eq!(session.retained_body_bytes, 0); + assert!( + session + .invocations + .iter() + .any(|invocation| invocation.outcome + == HttpResponseInvocationOutcome::FailOpen) + ); + // Holding returned output applies backpressure: subsequent + // stage work starts only when the relay calls again. + drop(output); + for _ in 0..3 { + let output = session.push_body(vec![1]).await.unwrap(); + assert!( + output.iter().map(Vec::len).sum::() + < MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES + ); + assert_eq!(session.retained_body_bytes, 0); + } + let finish = session.finish(Vec::new()).await.unwrap(); + assert!( + finish.body_units.iter().map(Vec::len).sum::() + < MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES + ); + } + Err(failure) => { + assert_eq!(on_error, OnError::FailClosed); + assert!( + failure + .reason + .contains("response_body_aggregate_over_capacity") + ); + session + .end(MiddlewareSessionEndReason::MiddlewareFailure) + .await; + } + } + assert_eq!( + runner.registry.session_admission.available_permits(), + permits + ); + } + } + #[tokio::test] async fn headers_only_preflight_applies_end_to_end_mutation() { let runner = ChainRunner::new(Arc::new(ResponseService { @@ -2651,8 +2988,9 @@ mod tests { assert!(outcome.session.is_none()); let _ = shutdown_tx.send(()); - server_task + tokio::time::timeout(Duration::from_secs(2), server_task) .await + .expect("bounded server shutdown") .expect("join response middleware server") .expect("serve response middleware"); } diff --git a/crates/openshell-supervisor-network/src/l7/rest.rs b/crates/openshell-supervisor-network/src/l7/rest.rs index ec0da7d13..e918a292c 100644 --- a/crates/openshell-supervisor-network/src/l7/rest.rs +++ b/crates/openshell-supervisor-network/src/l7/rest.rs @@ -3378,7 +3378,21 @@ where guard.ensure_current()?; } let header_bytes = &buffered[..header_end]; - let parsed = match parse_response_head_for_middleware(header_bytes) { + // Ordinary responses retain the HTTP parser's byte-preserving behavior. + // Response-specific normalization and limits apply only to selected hooks. + if middleware.chain.is_empty() { + return Ok(None); + } + let parsed = match middleware + .runner + .describe_http_response_chain(middleware.chain) + .await + { + Ok(described) if described.is_empty() => return Ok(None), + Ok(_) => parse_response_head_for_middleware(header_bytes), + Err(error) => Err(error), + }; + let parsed = match parsed { Ok(parsed) => parsed, Err(error) => { debug!(error = %error, "HTTP response head normalization failed"); @@ -4412,9 +4426,10 @@ where let length = usize::try_from(remaining) .unwrap_or(unit_limit) .min(unit_limit); - let unit = read_exact_response_with_deadline( + let unit = read_response_payload_with_deadline( reader, length, + !pending.is_empty(), session, client, &mut framing, @@ -4423,6 +4438,7 @@ where if let Some(guard) = generation_guard { guard.ensure_current()?; } + let partial = unit.len() < length; remaining -= unit.len() as u64; buffer_normalized_response_bytes( session, @@ -4433,6 +4449,15 @@ where unit_limit, ) .await?; + if partial { + flush_normalized_response_bytes( + session, + client, + std::mem::take(&mut pending), + &mut framing, + ) + .await?; + } } flush_normalized_response_bytes(session, client, pending, &mut framing).await?; Ok(Vec::new()) @@ -4464,9 +4489,10 @@ where let mut remaining = chunk_size; while remaining > 0 { let length = remaining.min(unit_limit); - let unit = read_exact_response_with_deadline( + let unit = read_response_payload_with_deadline( reader, length, + !pending.is_empty(), session, client, &mut framing, @@ -4475,6 +4501,7 @@ where if let Some(guard) = generation_guard { guard.ensure_current()?; } + let partial = unit.len() < length; remaining -= unit.len(); buffer_normalized_response_bytes( session, @@ -4485,19 +4512,37 @@ where unit_limit, ) .await?; + if partial { + flush_normalized_response_bytes( + session, + client, + std::mem::take(&mut pending), + &mut framing, + ) + .await?; + } } - if read_exact_response_with_deadline(reader, 2, session, client, &mut framing) - .await? - .as_slice() - != b"\r\n" + let terminator = if let Ok(result) = + tokio::time::timeout(RESPONSE_UNIT_COALESCE_TIMEOUT, reader.read_exact_vec(2)) + .await { + result? + } else { + flush_normalized_response_bytes( + session, + client, + std::mem::take(&mut pending), + &mut framing, + ) + .await?; + read_exact_response_with_deadline(reader, 2, session, client, &mut framing) + .await? + }; + if terminator != b"\r\n" { return Err(miette!("HTTP response chunk is missing its terminator")); } - size_line = if let Ok(line) = tokio::time::timeout( - RESPONSE_UNIT_COALESCE_TIMEOUT, - read_response_line_with_deadline(reader, session, client, &mut framing), - ) - .await + size_line = if let Ok(line) = + tokio::time::timeout(RESPONSE_UNIT_COALESCE_TIMEOUT, reader.read_line()).await { line? } else { @@ -4513,28 +4558,45 @@ where } } BodyLength::None if server_wants_close || event_stream => loop { - let read = - read_response_with_deadline(reader, unit_limit, session, client, &mut framing); - let next = if pending.is_empty() && event_stream { - read.await? - } else if pending.is_empty() { - match tokio::time::timeout(RELAY_EOF_IDLE_TIMEOUT, read).await { - Ok(result) => result?, - Err(_) => None, + // Cancel only input acquisition, never expiry or client writes. + let wait = if pending.is_empty() { + if event_stream { + None + } else { + Some(RELAY_EOF_IDLE_TIMEOUT) } - } else if let Ok(result) = - tokio::time::timeout(RESPONSE_UNIT_COALESCE_TIMEOUT, read).await - { - result? } else { - flush_normalized_response_bytes( - session, - client, - std::mem::take(&mut pending), - &mut framing, - ) - .await?; - continue; + Some(RESPONSE_UNIT_COALESCE_TIMEOUT) + }; + let read_deadline = wait.map(|wait| tokio::time::Instant::now() + wait); + let whole_deadline = session.whole_body_deadline(); + let deadline = match (read_deadline, whole_deadline) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + }; + let next = if let Some(deadline) = deadline { + if let Ok(result) = + tokio::time::timeout_at(deadline, reader.read_some(unit_limit)).await + { + result? + } else { + if whole_deadline.is_some_and(|d| d <= tokio::time::Instant::now()) { + expire_whole_body_deadline(session, client, &mut framing).await?; + } else if pending.is_empty() { + return Ok(Vec::new()); + } else { + flush_normalized_response_bytes( + session, + client, + std::mem::take(&mut pending), + &mut framing, + ) + .await?; + } + continue; + } + } else { + reader.read_some(unit_limit).await? }; let Some(unit) = next else { flush_normalized_response_bytes(session, client, pending, &mut framing).await?; @@ -4611,25 +4673,52 @@ async fn expire_whole_body_deadline( deliver_response_units(client, output, framing, session.requires_whole_body()).await } -async fn read_response_with_deadline( +// Unit size is a maximum. Coalesce available payload without waiting for +// the rest of a transfer chunk, and finish expiry outside read timeouts. +async fn read_response_payload_with_deadline( reader: &mut BufferedResponseReader<'_, R>, limit: usize, + has_pending: bool, session: &mut openshell_supervisor_middleware::HttpResponseSession, client: &mut C, framing: &mut ResponseOutputState<'_>, -) -> Result>> +) -> Result> where R: AsyncRead + Unpin, C: AsyncWrite + Unpin, { + let mut payload = Vec::new(); + let mut coalesce_deadline = + has_pending.then(|| tokio::time::Instant::now() + RESPONSE_UNIT_COALESCE_TIMEOUT); loop { - let Some(deadline) = session.whole_body_deadline() else { - return reader.read_some(limit).await; + let whole_deadline = session.whole_body_deadline(); + let deadline = match (coalesce_deadline, whole_deadline) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), }; - match tokio::time::timeout_at(deadline, reader.read_some(limit)).await { - Ok(result) => return result, - Err(_) => expire_whole_body_deadline(session, client, framing).await?, + let read = reader.read_some(limit - payload.len()); + let result = if let Some(deadline) = deadline { + if let Ok(result) = tokio::time::timeout_at(deadline, read).await { + result + } else { + if whole_deadline.is_some_and(|d| d <= tokio::time::Instant::now()) { + expire_whole_body_deadline(session, client, framing).await?; + } + if coalesce_deadline.is_some_and(|d| d <= tokio::time::Instant::now()) { + return Ok(payload); + } + continue; + } + } else { + read.await + }; + let data = result?.ok_or_else(|| miette!("HTTP response body ended unexpectedly"))?; + payload.extend_from_slice(&data); + if payload.len() == limit { + return Ok(payload); } + coalesce_deadline + .get_or_insert_with(|| tokio::time::Instant::now() + RESPONSE_UNIT_COALESCE_TIMEOUT); } } @@ -5253,12 +5342,14 @@ mod tests { BlockWholeBody, BlockStream, SlowWholeBody, + SlowStream, InvalidBodySequence, InvalidWholeBodySequence, } struct ResponseRelayService { script: ResponseRelayScript, + request_only: bool, } #[tonic::async_trait] @@ -5268,8 +5359,16 @@ mod tests { name: "test/response-relay".into(), service_version: "test".into(), bindings: vec![MiddlewareBinding { - operation: SupervisorMiddlewareOperation::HttpResponse as i32, - phase: SupervisorMiddlewarePhase::PreReturn as i32, + operation: if self.request_only { + SupervisorMiddlewareOperation::HttpRequest + } else { + SupervisorMiddlewareOperation::HttpResponse + } as i32, + phase: if self.request_only { + SupervisorMiddlewarePhase::PreCredentials + } else { + SupervisorMiddlewarePhase::PreReturn + } as i32, max_payload_bytes: 4096, timeout: String::new(), }], @@ -5302,7 +5401,11 @@ mod tests { openshell_supervisor_middleware::HttpResponseResultStream, tonic::Status, > { - let script = self.script; + assert!( + !self.request_only, + "request-only service received a response" + ); + let mut script = self.script; let (sender, receiver) = mpsc::channel(4); tokio::spawn(async move { while let Some(event) = requests.recv().await { @@ -5310,7 +5413,14 @@ mod tests { break; }; let result = match event { - http_response_event::Event::Preflight(_) => { + http_response_event::Event::Preflight(preflight) => { + if preflight + .config + .as_ref() + .is_some_and(|config| config.fields.contains_key("whole_body")) + { + script = ResponseRelayScript::WholeBody; + } if matches!(script, ResponseRelayScript::BlockPreflight) { HttpResponseEventResult { result: Some( @@ -5345,6 +5455,7 @@ mod tests { (HttpResponseBodyMode::WholeBodyBytes, Vec::new()) } ResponseRelayScript::Stream + | ResponseRelayScript::SlowStream | ResponseRelayScript::BlockStream | ResponseRelayScript::InvalidBodySequence => { (HttpResponseBodyMode::StreamBytes, Vec::new()) @@ -5384,6 +5495,7 @@ mod tests { [b"whole:".as_slice(), &data].concat() } ResponseRelayScript::Stream + | ResponseRelayScript::SlowStream | ResponseRelayScript::BlockStream | ResponseRelayScript::InvalidBodySequence => { data.to_ascii_uppercase() @@ -5391,7 +5503,11 @@ mod tests { ResponseRelayScript::HeadersOnly | ResponseRelayScript::BlockPreflight => break, }; - if matches!(script, ResponseRelayScript::SlowWholeBody) { + if matches!( + script, + ResponseRelayScript::SlowWholeBody + | ResponseRelayScript::SlowStream + ) { tokio::time::sleep(std::time::Duration::from_millis(75)).await; } HttpResponseEventResult { @@ -7326,6 +7442,132 @@ mod tests { assert!(!is_bodiless_response("POST", 201)); } + #[tokio::test] + async fn response_middleware_unbound_chains_preserve_ordinary_headers() { + let mut many_headers = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n".to_vec(); + for _ in 0..129 { + many_headers.extend_from_slice(b"Set-Cookie: a=b\r\n"); + } + many_headers.extend_from_slice(b"\r\n"); + let opaque_headers = + b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\nX-Opaque: \xff\xfe\r\n\r\nabc".to_vec(); + let (_, chain) = response_middleware_fixture(ResponseRelayScript::HeadersOnly); + let request_only_runner = + openshell_supervisor_middleware::ChainRunner::new(Arc::new(ResponseRelayService { + script: ResponseRelayScript::HeadersOnly, + request_only: true, + })); + let empty_runner = openshell_supervisor_middleware::ChainRunner::default(); + for response in [many_headers, opaque_headers] { + assert!(response.len() < MAX_HEADER_BYTES); + for context in [ + None, + Some(response_middleware_context(&empty_runner, &[], "GET")), + Some(response_middleware_context( + &request_only_runner, + &chain, + "GET", + )), + ] { + let mut upstream = response.as_slice(); + let mut delivered = Vec::new(); + let outcome = relay_response( + "GET", + &mut upstream, + &mut delivered, + RelayResponseOptions::default(), + context, + ) + .await + .unwrap(); + assert!(matches!(outcome, RelayOutcome::Reusable)); + assert_eq!(delivered, response); + } + } + } + + #[tokio::test] + async fn response_middleware_selected_hook_enforces_header_limits() { + let mut response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n".to_vec(); + for _ in 0..129 { + response.extend_from_slice(b"Set-Cookie: a=b\r\n"); + } + response.extend_from_slice(b"\r\n"); + let (runner, chain) = response_middleware_fixture(ResponseRelayScript::HeadersOnly); + let mut upstream = response.as_slice(); + let mut delivered = Vec::new(); + let outcome = relay_response( + "GET", + &mut upstream, + &mut delivered, + RelayResponseOptions::default(), + Some(response_middleware_context(&runner, &chain, "GET")), + ) + .await + .unwrap(); + assert!(matches!(outcome, RelayOutcome::Consumed)); + assert!(delivered.starts_with(b"HTTP/1.1 502 ")); + } + + #[tokio::test] + async fn response_middleware_flushes_partial_framed_payload_promptly() { + for chunked in [false, true] { + let (runner, chain) = response_middleware_fixture(ResponseRelayScript::Stream); + let (mut upstream_read, mut upstream_write) = tokio::io::duplex(8192); + // Small capacity forces the relay to complete partial downstream writes. + let (mut client_read, mut client_write) = tokio::io::duplex(7); + let head = if chunked { + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nContent-Type: text/event-stream\r\n\r\n1000\r\nabc".as_slice() + } else { + b"HTTP/1.1 200 OK\r\nContent-Length: 4096\r\nContent-Type: text/event-stream\r\n\r\nabc".as_slice() + }; + upstream_write.write_all(head).await.unwrap(); + let task = tokio::spawn(async move { + relay_response( + "GET", + &mut upstream_read, + &mut client_write, + RelayResponseOptions::default(), + Some(response_middleware_context(&runner, &chain, "GET")), + ) + .await + }); + let result = tokio::time::timeout(std::time::Duration::from_secs(2), async { + let mut head = Vec::new(); + while !head.ends_with(b"\r\n\r\n") { + head.push(client_read.read_u8().await.unwrap()); + } + let mut first = [0; 8]; + client_read.read_exact(&mut first).await.unwrap(); + assert_eq!(&first, b"3\r\nABC\r\n"); + // Complete the same framed payload only after its first transformed + // bytes have reached the consumer. + upstream_write.write_all(&vec![b'd'; 4093]).await.unwrap(); + if chunked { + for fragment in [b"\r".as_slice(), b"\n0\r", b"\n\r", b"\n"] { + upstream_write.write_all(fragment).await.unwrap(); + tokio::task::yield_now().await; + } + } + drop(upstream_write); + let mut rest = Vec::new(); + client_read.read_to_end(&mut rest).await.unwrap(); + assert!(rest.ends_with(b"0\r\n\r\n")); + let body = collect_chunked_body(&mut tokio::io::empty(), &rest, None, None) + .await + .unwrap(); + assert_eq!(body, vec![b'D'; 4093]); + }) + .await; + if result.is_err() { + task.abort(); + } + let relay = task.await; + assert!(result.is_ok(), "partial payload stalled, chunked={chunked}"); + assert!(relay.unwrap().is_ok()); + } + } + fn response_middleware_fixture( script: ResponseRelayScript, ) -> ( @@ -7348,6 +7590,7 @@ mod tests { let runner = openshell_supervisor_middleware::ChainRunner::new(Arc::new(ResponseRelayService { script, + request_only: false, })); let chain = vec![openshell_supervisor_middleware::ChainEntry { name: "response".into(), @@ -7668,13 +7911,117 @@ mod tests { assert!(matches!(outcome.unwrap(), RelayOutcome::Reusable)); let delivered = String::from_utf8(delivered).unwrap(); - assert!( - delivered.ends_with("5\r\nhello\r\n0\r\n\r\n"), - "{delivered}" - ); + let (_, body) = delivered.split_once("\r\n\r\n").unwrap(); + let decoded = collect_chunked_body(&mut tokio::io::empty(), body.as_bytes(), None, None) + .await + .unwrap(); + assert_eq!(decoded, b"hello"); assert!(!delivered.contains("whole:hello"), "{delivered}"); } + #[tokio::test(start_paused = true)] + async fn response_middleware_expiry_preserves_bytes_through_slow_stream_and_client() { + for chunked in [true, false] { + let (runner, mut chain) = response_middleware_fixture_with_error( + ResponseRelayScript::SlowStream, + openshell_supervisor_middleware::OnError::FailOpen, + ); + let mut whole_body = chain[0].clone(); + whole_body.name = "whole-body".into(); + whole_body + .config + .fields + .insert("whole_body".into(), prost_types::Value::default()); + chain[0].order = 1; + chain.insert(0, whole_body); + let (mut upstream_read, mut upstream_write) = tokio::io::duplex(8192); + // Force write_all to make partial progress before each wait. + let (mut client_read, mut client_write) = tokio::io::duplex(7); + let producer = async move { + let head = if chunked { + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n800\r\n".as_slice() + } else { + b"HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n".as_slice() + }; + upstream_write.write_all(head).await.unwrap(); + upstream_write.write_all(&vec![b'a'; 2048]).await.unwrap(); + if chunked { + upstream_write.write_all(b"\r\n").await.unwrap(); + } + // The first coalesced unit belongs to the whole-body stage. A new + // partial unit starts coalescing just before its deadline. + tokio::time::sleep(std::time::Duration::from_millis(9)).await; + upstream_write + .write_all(if chunked { b"1\r\nb\r\n" } else { b"b" }) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + if chunked { + upstream_write.write_all(b"0\r\n\r\n").await.unwrap(); + } + upstream_write.shutdown().await.unwrap(); + }; + let relay = async { + let mut context = response_middleware_context(&runner, &chain, "GET"); + context.whole_body_timeout = std::time::Duration::from_millis(10); + let outcome = relay_response( + "GET", + &mut upstream_read, + &mut client_write, + RelayResponseOptions::default(), + Some(context), + ) + .await; + drop(client_write); + outcome + }; + let consumer = async move { + let mut delivered = Vec::new(); + let mut bytes = [0; 7]; + loop { + let count = client_read.read(&mut bytes).await.unwrap(); + if count == 0 { + break; + } + delivered.extend_from_slice(&bytes[..count]); + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + } + delivered + }; + let ((), outcome, delivered) = tokio::time::timeout( + std::time::Duration::from_secs(3), + Box::pin(async { tokio::join!(producer, relay, consumer) }), + ) + .await + .expect("response relay stalled"); + assert!(outcome.is_ok(), "{outcome:?}"); + assert!(delivered.starts_with(b"HTTP/1.1 200 OK\r\n")); + let head_end = delivered + .windows(4) + .position(|bytes| bytes == b"\r\n\r\n") + .unwrap() + + 4; + let mut wire = &delivered[head_end..]; + let mut body = Vec::new(); + loop { + let end = wire.windows(2).position(|bytes| bytes == b"\r\n").unwrap(); + let size = + usize::from_str_radix(std::str::from_utf8(&wire[..end]).unwrap(), 16).unwrap(); + wire = &wire[end + 2..]; + if size == 0 { + assert_eq!(wire, b"\r\n"); + break; + } + body.extend_from_slice(&wire[..size]); + assert_eq!(&wire[size..size + 2], b"\r\n"); + wire = &wire[size + 2..]; + } + let mut expected = vec![b'A'; 2048]; + expected.push(b'B'); + assert_eq!(body, expected, "chunked={chunked}"); + } + } + #[tokio::test] async fn response_middleware_streams_normalized_chunks_and_preserves_trailers() { let (outcome, delivered) = run_response_middleware_relay( diff --git a/docs/reference/gateway-config.mdx b/docs/reference/gateway-config.mdx index 8b1cb5c58..878a6665e 100644 --- a/docs/reference/gateway-config.mdx +++ b/docs/reference/gateway-config.mdx @@ -315,6 +315,8 @@ The gateway connects to every registered service and validates `Describe` before `timeout` is the operator-configured service-wide RPC timeout. It accepts the same compact duration syntax as gateway interceptors: an integer followed by `ms` or `s`, such as `500ms` or `2s`. Values must be between `10ms` and `30s`, inclusive. Omit the field to use the 500 ms platform default. A binding may advertise a shorter `timeout` in the `Describe` manifest, but it cannot extend the operator-configured deadline; OpenShell uses the smaller value. OpenShell validates both levels before accepting the service. The operator-configured service timeout applies to `Describe` and `ValidateConfig`. The effective binding timeout applies to HTTP request evaluation, HTTP response preflight and unit exchanges, WebSocket preflight, and each WebSocket message. Accepted streaming protocols have no connection-wide RPC deadline. +Response middleware also has an 8 MiB aggregate retained-body limit per response session across stage buffers and pending output. This limit is separate from each service's `max_payload_bytes` and cannot be configured. A replacement that exceeds the available budget follows the stage's `on_error` policy. Fail-open disables that stage and forwards its original input; fail-closed returns 502 before commitment or stops delivery after commitment. Streaming inputs flush after a short coalescing window without waiting for an entire upstream transfer chunk. + `http_response_whole_body_timeout` is a supervisor-wide safety bound, not a middleware RPC timeout. It defaults to `120s` and accepts a positive integer followed by `ms`, `s`, or `m`. The timer starts after ordered response preflight selects a `WHOLE_BODY_BYTES` stage and before OpenShell reads the first response body byte. It does not reset as chunks arrive. It ends after the normalized body reaches end of stream and every whole-body barrier produces output. On expiry, each active whole-body stage follows its configured `on_error` policy: fail-open releases OpenShell-owned bytes through later stages, while fail-closed returns the canonical pre-commit 502 response. Changing this field requires restarting the gateway. The service `grpc_endpoint` supports plaintext `http://` and TLS `https://`. HTTPS uses the platform trust store unless `tls_ca_cert_path` names a certificate-only PEM bundle. OpenShell rejects bundles containing private keys, loads the certificates at gateway startup, and distributes only public certificates to sandbox supervisors; normal TLS hostname verification still applies. `audience` sets the exact audience for gateway-minted service tokens and defaults to `urn:openshell:extension:middleware:`. After authenticated `Describe` succeeds, OpenShell treats a non-empty manifest `expected_audience` as a consistency assertion and refuses to start when it differs from the configured audience. A strict verifier may reject an incorrect audience before returning the manifest. diff --git a/examples/supervisor-middleware-content-guard/README.md b/examples/supervisor-middleware-content-guard/README.md index 4ccda9162..c9f4a6f92 100644 --- a/examples/supervisor-middleware-content-guard/README.md +++ b/examples/supervisor-middleware-content-guard/README.md @@ -15,7 +15,7 @@ This configured-literal guard applies the same case-sensitive terms to UTF-8 HTT ## Prerequisites -Install `cargo`, `curl`, `jq`, `openssl`, and `uv` with Python 3 on the host before running the smoke script. +Install `cargo`, `curl`, `jq`, `openssl`, `mise`, and `uv` with Python 3 on the host before running the smoke script. Start Docker or Podman. The supervisor image build uses the repository's Linux cross-compilation toolchain, including `cargo-zigbuild` and Zig on macOS. Install the repository's mise tools before running it. ## Run the smoke example @@ -39,7 +39,15 @@ The script creates the sandbox and prints the guarded and unguarded request comm CONTENT_GUARD_SMOKE_HOST=192.168.1.10 ./examples/supervisor-middleware-content-guard/smoke.sh --test-suite ``` -The gateway auto-detects its compute driver. Set `CONTENT_GUARD_SMOKE_DRIVER=docker` or `CONTENT_GUARD_SMOKE_DRIVER=podman` if more than one local runtime is installed and auto-detection selects the wrong one. +The script defaults to Docker. Set `CONTENT_GUARD_SMOKE_DRIVER=podman` to build and run with Podman instead. + +On Linux and macOS, the script runs `mise run docker:build:supervisor` with the selected container engine to build a Linux supervisor from the current checkout. It configures that driver's `supervisor_image` with a unique local tag, so the response checks exercise the local runtime changes. macOS host binaries are never used inside the sandbox. The local image remains available after the smoke run. + +Cargo's configured target directory applies to the host binaries and the Linux supervisor build. For example: + +```shell +CARGO_TARGET_DIR=/tmp/content-guard-target ./examples/supervisor-middleware-content-guard/smoke.sh --test-suite +``` ## Run manually diff --git a/examples/supervisor-middleware-content-guard/smoke.sh b/examples/supervisor-middleware-content-guard/smoke.sh index a372a7ffe..f16ed28ca 100755 --- a/examples/supervisor-middleware-content-guard/smoke.sh +++ b/examples/supervisor-middleware-content-guard/smoke.sh @@ -25,7 +25,7 @@ Environment: CONTENT_GUARD_SMOKE_HOST Non-loopback host address reachable from both the gateway and sandbox containers. CONTENT_GUARD_SMOKE_DRIVER - Optional compute driver name, such as docker or podman. + Compute driver: docker (default) or podman. EOF } @@ -105,7 +105,14 @@ detect_service_host() { } SERVICE_HOST="$(detect_service_host)" -COMPUTE_DRIVER="${CONTENT_GUARD_SMOKE_DRIVER:-}" +COMPUTE_DRIVER="${CONTENT_GUARD_SMOKE_DRIVER:-docker}" +case "$COMPUTE_DRIVER" in + docker | podman) ;; + *) + echo "CONTENT_GUARD_SMOKE_DRIVER must be docker or podman" >&2 + exit 1 + ;; +esac if [[ "$SERVICE_HOST" == "localhost" || "$SERVICE_HOST" == "::1" || "$SERVICE_HOST" == 127.* || "$SERVICE_HOST" == *:* ]]; then echo "CONTENT_GUARD_SMOKE_HOST must be a non-loopback IPv4 address: $SERVICE_HOST" >&2 exit 1 @@ -121,6 +128,7 @@ MIDDLEWARE_LOG="$LOG_DIR/middleware.log" UPSTREAM_LOG="$LOG_DIR/upstream.log" SANDBOX_LOG="$LOG_DIR/sandbox.log" RUN_ID="content-guard-smoke-$$-$RANDOM" +SUPERVISOR_IMAGE="localhost/openshell-content-guard/supervisor:$RUN_ID" # Sandbox names are capped at 19 characters. Use a short prefix with # the PID for uniqueness; keep the full RUN_ID for gateway identity. SANDBOX_NAME="cg-$$-$RANDOM" @@ -228,8 +236,8 @@ allow_insecure_transport = true max_payload_bytes = 262144 timeout = "500ms" -[openshell.drivers.docker] -supervisor_bin = "$ROOT/target/debug/openshell-sandbox" +[openshell.drivers.$COMPUTE_DRIVER] +supervisor_image = "$SUPERVISOR_IMAGE" EOF } @@ -533,13 +541,21 @@ require_command curl require_command jq require_command openssl require_command uv +require_command mise ROOT_TARGET_DIR="$(cargo_target_dir "$ROOT/Cargo.toml")" EXAMPLE_TARGET_DIR="$(cargo_target_dir "$EXAMPLE_DIR/Cargo.toml")" GATEWAY_BIN="$ROOT_TARGET_DIR/debug/openshell-gateway" CLI_BIN="$ROOT_TARGET_DIR/debug/openshell" MIDDLEWARE_BIN="$EXAMPLE_TARGET_DIR/debug/supervisor-middleware-content-guard" run_setup_step "building gateway" cargo build --quiet -p openshell-gateway --bin openshell-gateway -run_setup_step "building sandbox supervisor" cargo build --quiet -p openshell-sandbox --bin openshell-sandbox +# Always rebuild from this checkout and load into the selected runtime. Native +# macOS binaries cannot run in Linux sandboxes; Podman also needs an image. +# A unique tag prevents the driver from selecting an older published runtime. +run_setup_step "building Linux sandbox supervisor image" \ + env -u CI -u DOCKER_PLATFORM -u DOCKER_PUSH -u DOCKER_OUTPUT \ + CONTAINER_ENGINE="$COMPUTE_DRIVER" PREBUILT_AUTO_STAGE=1 \ + IMAGE_REGISTRY=localhost/openshell-content-guard IMAGE_TAG="$RUN_ID" \ + mise run docker:build:supervisor run_setup_step "building content guard" cargo build --quiet --manifest-path "$EXAMPLE_DIR/Cargo.toml" run_setup_step "building CLI" cargo build --quiet -p openshell-cli --bin openshell generate_gateway_jwt_bundle diff --git a/skills/generate-sandbox-policy/SKILL.md b/skills/generate-sandbox-policy/SKILL.md index 2f7436e56..26cfd736d 100644 --- a/skills/generate-sandbox-policy/SKILL.md +++ b/skills/generate-sandbox-policy/SKILL.md @@ -214,7 +214,7 @@ Add `network_middlewares` only when the user asks to inspect, transform, redact, - Use `openshell/regex` without gateway registration for fixed-pattern redaction of UTF-8 HTTP request bodies or complete client-to-upstream WebSocket text messages. - Use an operator-owned middleware name only when it is already registered under `[[openshell.supervisor.middleware]]` and reachable from both the gateway and sandbox supervisors. - Confirm that a requested WebSocket implementation exposes a `WEBSOCKET_MESSAGE/PRE_CREDENTIALS` binding. `openshell/regex` exposes this binding. A host-matched HTTP-only implementation may inspect the upgrade GET but does not join the post-upgrade chain; messages pass and OpenShell emits `binding_not_selected` coverage regardless of `on_error`. -- Confirm that requested response processing exposes `HTTP_RESPONSE/PRE_RETURN`. Response stages choose header-only, whole-body, or streaming inspection independently. Whole-body stages delay response commitment and are bounded by the gateway's supervisor-wide accumulation timeout. Intentional blocks produce a canonical 403 before commitment and abort the connection after commitment. +- Confirm that requested response processing exposes `HTTP_RESPONSE/PRE_RETURN`. Response stages choose header-only, whole-body, or streaming inspection independently. Whole-body stages delay response commitment and are bounded by the gateway's supervisor-wide accumulation timeout. Expanding response transformations also share an 8 MiB retained-body budget per session; exhaustion follows the stage's `on_error` policy. Intentional blocks produce a canonical 403 before commitment and abort the connection after commitment. - WebSocket middleware runs for both `ws://` and `wss://` and receives complete client text messages only. Binary messages pass under both error modes and emit `unsupported_message_type` coverage for active stages. Upstream-to-client messages remain uninspected. Do not claim that V1 provides all-message WebSocket inspection. - Treat `fail_open` on WebSocket as a session-scoped bypass: if the stage stream fails, OpenShell disables it for later messages on that connection and emits a state-change finding. Prefer `fail_closed` for required redaction or authorization. - `on_error` governs failures after an advertised operation binding is selected. It does not apply to an unadvertised WebSocket binding or binary-message pass-through. An explicit HTTP, WebSocket preflight, or WebSocket message denial is authoritative under both `fail_open` and `fail_closed`. diff --git a/tasks/scripts/stage-prebuilt-binaries.sh b/tasks/scripts/stage-prebuilt-binaries.sh index fe4913439..ff7eda26f 100755 --- a/tasks/scripts/stage-prebuilt-binaries.sh +++ b/tasks/scripts/stage-prebuilt-binaries.sh @@ -171,6 +171,7 @@ build_component_for_arch() { local current_host_os local current_host_arch local binary_path + local cargo_output_dir local build_rustflags resolve_component "$component" @@ -257,7 +258,8 @@ build_component_for_arch() { CARGO_INCREMENTAL=0 mise x -- "${cargo_env[@]}" "${cargo_subcommand[@]}" "${args[@]}" ) - binary_path="${ROOT}/target/${target}/release/${binary}" + cargo_output_dir="$(cd "$ROOT" && mise x -- cargo metadata --format-version=1 --no-deps | jq -er '.target_directory')" + binary_path="${cargo_output_dir}/${target}/release/${binary}" if [[ "$component" == "gateway" ]]; then "$SCRIPT_DIR/verify-glibc-symbols.sh" 2.28 "$binary_path" elif [[ "$component" == "supervisor" ]]; then