From af45186b57f010df49442ec8d872f7d2ff7c5de0 Mon Sep 17 00:00:00 2001 From: Shiju Date: Fri, 6 Nov 2026 20:20:39 +0530 Subject: [PATCH 1/4] fix(proxy): handle upstream HTTP connection closure Signed-off-by: Shiju --- .../src/l7/relay.rs | 89 +++++- .../src/l7/relay/upstream_close_tests.rs | 294 ++++++++++++++++++ .../src/l7/rest.rs | 134 ++++++-- .../src/l7/rest/http_response.rs | 4 +- docs/how-it-works/policies/network-rules.mdx | 7 + 5 files changed, 496 insertions(+), 32 deletions(-) create mode 100644 crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index b4acb5e52f..671f739e3c 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -33,11 +33,13 @@ use openshell_ocsf::{ use std::collections::BTreeMap; use std::collections::HashSet; use std::sync::Arc; -use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tracing::{debug, warn}; #[cfg(test)] mod token_grant_ownership_tests; +#[cfg(test)] +mod upstream_close_tests; const CONNECTION_READ_AHEAD_BYTES: usize = 8 * 1024; @@ -814,6 +816,27 @@ where .await } +fn build_upstream_http_failure_event( + ctx: &L7EvalContext, + method: &str, + response_started: bool, +) -> openshell_ocsf::OcsfEvent { + let mut event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::for_http_method(method)) + .http_request(method_only_request(method)) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(SeverityId::Low) + .status(StatusId::Failure) + .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) + .message("HTTP upstream connection closed or failed") + .status_detail("upstream_http_transport_failure"); + if !response_started { + event = event.http_response(openshell_ocsf::HttpResponse { code: 502 }); + } + event.build() +} + async fn relay_http_request_with_credential_rejection_observed( request: &crate::l7::provider::L7Request, client: &mut C, @@ -841,7 +864,20 @@ where { Ok(outcome) => Ok(Some(outcome)), Err(report) => { - if let Some(error) = report.downcast_ref::() { + if let Some(error) = report.downcast_ref::() { + ocsf_emit!(build_upstream_http_failure_event( + ctx, + &request.action, + error.response_started, + )); + // A missing response does not prove the request was not executed. + // Never replay it, and never append a 502 to a started response. + if !error.response_started { + write_bad_gateway_response(client, None).await?; + } + Ok(None) + } else if let Some(error) = report.downcast_ref::() + { // Body classification includes credentials for other destinations, // so denial does not establish that this endpoint lacks credentials. // Record the local decision before cleanup or client I/O can fail. @@ -1211,7 +1247,7 @@ where .endpoint_observation_tx .as_ref() .and_then(EndpointObservationSender::capture); - let mut req = match provider.parse_request(client).await { + let mut req = match parse_next_http_request(&provider, client, upstream, ctx).await { Ok(Some(req)) => req, Ok(None) => return Ok(()), Err(e) => { @@ -2077,7 +2113,7 @@ where .filter(|_| ctx.token_grant_resolver.is_some()) .and_then(EndpointObservationSender::capture); // Parse one HTTP request from client - let req = match provider.parse_request(client).await { + let req = match parse_next_http_request(&provider, client, upstream, ctx).await { Ok(Some(req)) => req, Ok(None) => return Ok(()), // Client closed connection Err(e) => { @@ -2774,7 +2810,7 @@ where // Validate the head, including body framing, before deciding whether // this endpoint can inspect the requested protocol. Upgrade refusal // must not wait for a body or depend on its inspection size limit. - let mut req = match provider.parse_request(client).await { + let mut req = match parse_next_http_request(&provider, client, upstream, ctx).await { Ok(Some(req)) => req, Ok(None) => return Ok(()), Err(e) => { @@ -3528,6 +3564,47 @@ fn evaluate_l7_request_once( Ok((allowed, reason)) } +/// Read a request while watching the otherwise idle upstream connection. +/// +/// A closed upstream must not leave a live socket in the client's pool. If +/// both sides are ready, finish the request first so a forwarding failure can +/// produce an explicit HTTP error. Unsolicited upstream bytes are not a response +/// to a request and cannot be assigned to the next exchange. +async fn parse_next_http_request( + provider: &crate::l7::rest::RestProvider, + client: &mut C, + upstream: &mut U, + ctx: &L7EvalContext, +) -> Result> +where + C: AsyncRead + AsyncWrite + Unpin + Send, + U: AsyncRead + Unpin, +{ + let mut byte = [0u8; 1]; + tokio::select! { + biased; + request = provider.parse_request(client) => request, + result = upstream.read(&mut byte) => { + let (detail, status, severity) = match result { + Ok(0) => ("HTTP upstream closed while idle", StatusId::Success, SeverityId::Informational), + Ok(_) => ("HTTP upstream sent unsolicited data while idle", StatusId::Failure, SeverityId::Low), + Err(_) => ("HTTP upstream failed while idle", StatusId::Failure, SeverityId::Low), + }; + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Close) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(severity) + .status(status) + .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) + .message(detail) + .build(); + ocsf_emit!(event); + Ok(None) + } + } +} + /// Relay HTTP traffic with credential injection only (no L7 OPA evaluation). /// /// Used when TLS is auto-terminated but no L7 policy (`protocol` + `access`/`rules`) @@ -3562,7 +3639,7 @@ where .filter(|_| ctx.token_grant_resolver.is_some()) .and_then(EndpointObservationSender::capture); // Read next request from client. - let req = match provider.parse_request(client).await { + let req = match parse_next_http_request(&provider, client, upstream, ctx).await { Ok(Some(req)) => req, Ok(None) => break, // Client closed connection. Err(e) => { diff --git a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs new file mode 100644 index 0000000000..8924b717fb --- /dev/null +++ b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs @@ -0,0 +1,294 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::l7::provider::{BodyLength, L7Request}; +use crate::opa::{NetworkInput, OpaEngine}; +use std::collections::HashMap; +use std::path::PathBuf; +use std::time::Duration; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::timeout; + +const REQUEST: &[u8] = + b"POST /mcp HTTP/1.1\r\nHost: api.example.test\r\nContent-Length: 2\r\n\r\n{}"; +const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"; +const BAD_GATEWAY: &[u8] = + b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + +fn context() -> L7EvalContext { + L7EvalContext { + host: "api.example.test".into(), + port: 80, + request_default_port: Some(80), + policy_name: "rest_api".into(), + binary_path: "/usr/bin/curl".into(), + ..Default::default() + } +} + +async fn read_exact_message(stream: &mut S, expected: &[u8]) { + let mut bytes = vec![0; expected.len()]; + stream.read_exact(&mut bytes).await.unwrap(); + assert_eq!(bytes, expected); +} + +async fn tcp_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = TcpStream::connect(listener.local_addr().unwrap()) + .await + .unwrap(); + let (server, _) = listener.accept().await.unwrap(); + openshell_core::net::set_tcp_nodelay_best_effort(&client); + openshell_core::net::set_tcp_nodelay_best_effort(&server); + (client, server) +} + +#[tokio::test] +async fn keep_alive_propagates_upstream_idle_close_over_tcp() { + timeout(Duration::from_secs(5), async { + for inspected in [false, true] { + let engine = OpaEngine::from_strings( + include_str!("../../../data/sandbox-policy.rego"), + r" +network_policies: + rest_api: + name: rest_api + endpoints: + - host: api.example.test + port: 80 + protocol: rest + enforcement: enforce + rules: + - allow: { method: POST, path: /mcp } + binaries: + - { path: /usr/bin/curl } +", + ) + .unwrap(); + let (mut app, mut client) = tcp_pair().await; + let (mut relay_upstream, mut upstream) = tcp_pair().await; + let relay = tokio::spawn(async move { + let ctx = context(); + if inspected { + let input = NetworkInput { + host: ctx.host.clone(), + port: ctx.port, + binary_path: PathBuf::from(&ctx.binary_path), + binary_sha256: "unused".into(), + ancestors: vec![], + cmdline_paths: vec![], + }; + let (config, generation) = engine + .query_endpoint_config_with_generation(&input) + .unwrap(); + let config = crate::l7::parse_l7_config(&config.unwrap()).unwrap(); + let evaluator = engine.clone_engine_for_tunnel(generation).unwrap(); + relay_with_inspection( + &config, + evaluator, + &mut client, + &mut relay_upstream, + &ctx, + ) + .await + } else { + let guard = engine + .generation_guard(engine.current_generation()) + .unwrap(); + relay_passthrough_with_credentials( + &mut client, + &mut relay_upstream, + &ctx, + &guard, + Some(&engine), + ) + .await + } + }); + // Healthy keep-alive must still carry more than one complete exchange. + for _ in 0..2 { + app.write_all(REQUEST).await.unwrap(); + read_exact_message(&mut upstream, REQUEST).await; + upstream.write_all(RESPONSE).await.unwrap(); + read_exact_message(&mut app, RESPONSE).await; + } + upstream.shutdown().await.unwrap(); + // Do not send another request: FIN must propagate while the client is idle. + let mut rest = Vec::new(); + app.read_to_end(&mut rest).await.unwrap(); + assert!( + rest.is_empty(), + "idle close must not invent an HTTP response" + ); + relay.await.unwrap().unwrap(); + } + }) + .await + .expect("idle upstream close must reach the client promptly"); +} + +#[tokio::test] +async fn queued_request_after_upstream_close_returns_502() { + timeout(Duration::from_secs(5), async { + let engine = OpaEngine::from_strings( + include_str!("../../../data/sandbox-policy.rego"), + "network_policies: {}\n", + ) + .unwrap(); + let (mut app, mut client) = tokio::io::duplex(8192); + let (mut relay_upstream, mut upstream) = tokio::io::duplex(8192); + // Both requests are buffered before the relay starts. When the first + // response arrives, request two and the upstream close are ready together. + app.write_all(&[REQUEST, REQUEST].concat()).await.unwrap(); + let relay = tokio::spawn(async move { + let guard = engine + .generation_guard(engine.current_generation()) + .unwrap(); + relay_passthrough_with_credentials( + &mut client, + &mut relay_upstream, + &context(), + &guard, + Some(&engine), + ) + .await + }); + read_exact_message(&mut upstream, REQUEST).await; + upstream.write_all(RESPONSE).await.unwrap(); + drop(upstream); + let mut received = Vec::new(); + app.read_to_end(&mut received).await.unwrap(); + assert_eq!(received, [RESPONSE, BAD_GATEWAY].concat()); + relay.await.unwrap().unwrap(); + }) + .await + .expect("queued request must receive an explicit gateway error"); +} + +#[tokio::test] +async fn upstream_failure_returns_502_only_before_response_starts() { + timeout(Duration::from_secs(5), async { + for response in [ + None, // Upstream disappeared before the first write. + Some(&b""[..]), // Request was accepted, then upstream closed without responding. + Some(&b"HTTP/1.1 100 Continue\r\n\r\n"[..]), + Some(&b"HTTP/1.1 200"[..]), + Some(&b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nOK"[..]), + Some(&b"HTTP/1.0 200 OK\r\n\r\nOK"[..]), // Valid close-delimited response. + ] { + let (mut app, mut client) = tokio::io::duplex(8192); + let (mut relay_upstream, mut upstream) = tokio::io::duplex(8192); + let server = response.map(|response| { + tokio::spawn(async move { + read_exact_message(&mut upstream, REQUEST).await; + upstream.write_all(response).await.unwrap(); + upstream.shutdown().await.unwrap(); + let mut extra = Vec::new(); + upstream.read_to_end(&mut extra).await.unwrap(); + assert!(extra.is_empty(), "POST must not be replayed"); + }) + }); + let relay = tokio::spawn(async move { + let req = L7Request { + action: "POST".into(), + target: "/mcp".into(), + query_params: HashMap::new(), + raw_header: REQUEST.to_vec(), + body_length: BodyLength::ContentLength(2), + }; + relay_http_request_with_credential_rejection( + &req, + &mut client, + &mut relay_upstream, + crate::l7::rest::RelayRequestOptions::default(), + &context(), + None, + ) + .await + }); + let mut received = Vec::new(); + app.read_to_end(&mut received).await.unwrap(); + let result = relay.await.unwrap(); + if response.is_none_or(<[u8]>::is_empty) { + assert_eq!(received, BAD_GATEWAY); + assert!(result.unwrap().is_none()); + } else { + let expected = if response == Some(&b"HTTP/1.0 200 OK\r\n\r\nOK"[..]) { + &b"HTTP/1.0 200 OK\r\nConnection: close\r\n\r\nOK"[..] + } else { + response.unwrap() + }; + assert_eq!(received, expected); + assert!( + !received + .windows(15) + .any(|bytes| bytes == b"502 Bad Gateway") + ); + } + if let Some(server) = server { + server.await.unwrap(); + } + } + }) + .await + .expect("failed upstream exchange must close promptly"); +} + +#[tokio::test] +async fn client_body_eof_is_not_an_upstream_failure() { + timeout(Duration::from_secs(5), async { + let (mut app, mut client) = tokio::io::duplex(8192); + let (mut relay_upstream, _upstream) = tokio::io::duplex(8192); + app.shutdown().await.unwrap(); + let req = L7Request { + action: "POST".into(), + target: "/mcp".into(), + query_params: HashMap::new(), + raw_header: REQUEST[..REQUEST.len() - 2].to_vec(), + body_length: BodyLength::ContentLength(2), + }; + let error = relay_http_request_with_credential_rejection( + &req, + &mut client, + &mut relay_upstream, + crate::l7::rest::RelayRequestOptions::default(), + &context(), + None, + ) + .await + .expect_err("client did not provide its declared body"); + assert!( + error + .downcast_ref::() + .is_none() + ); + drop(client); + let mut response = Vec::new(); + app.read_to_end(&mut response).await.unwrap(); + assert!( + response.is_empty(), + "client error must not be reported as a 502" + ); + }) + .await + .expect("client EOF must terminate the exchange"); +} + +#[test] +fn upstream_failure_event_reports_502_without_request_secrets() { + for response_started in [false, true] { + let event = build_upstream_http_failure_event(&context(), "POST", response_started); + let json = event.to_json().unwrap(); + assert_eq!(json["class_uid"], 4002); + assert_eq!(json["status_detail"], "upstream_http_transport_failure"); + assert_eq!(json["http_request"]["http_method"], "POST"); + assert!(json["http_request"].get("url").is_none()); + if response_started { + assert!(json.get("http_response").is_none()); + } else { + assert_eq!(json["http_response"]["code"], 502); + } + openshell_ocsf_validate::validate_event(&json).unwrap(); + } +} diff --git a/crates/openshell-supervisor-network/src/l7/rest.rs b/crates/openshell-supervisor-network/src/l7/rest.rs index 7a6a0d6236..0013063067 100644 --- a/crates/openshell-supervisor-network/src/l7/rest.rs +++ b/crates/openshell-supervisor-network/src/l7/rest.rs @@ -830,6 +830,21 @@ impl std::error::Error for CredentialUnavailableError {} impl miette::Diagnostic for CredentialUnavailableError {} +/// An upstream transport failure, distinct from client I/O and credential refusals. +#[derive(Debug)] +pub(crate) struct UpstreamHttpError { + pub(crate) response_started: bool, +} + +impl fmt::Display for UpstreamHttpError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("HTTP upstream connection closed or failed") + } +} + +impl std::error::Error for UpstreamHttpError {} +impl miette::Diagnostic for UpstreamHttpError {} + impl<'a> CredentialGenerationGuard<'a> { pub(crate) fn new( state: &'a openshell_core::provider_credentials::ProviderCredentialState, @@ -901,8 +916,43 @@ where C: AsyncRead + AsyncWrite + Unpin, U: AsyncRead + AsyncWrite + Unpin, { - let mut observed_upstream = ObservedUpstream { upstream, observer }; - let upstream = &mut observed_upstream; + let mut observed_upstream = ObservedUpstream { + upstream, + observer, + response_started: false, + transport_failed: false, + }; + let result = relay_http_exchange( + req, + client, + &mut observed_upstream, + options, + response_middleware, + observer, + ) + .await; + // IntoDiagnostic hides the concrete I/O error. Classify at the adapter's + // owner, where an upstream failure remains distinct from client I/O. + if result.is_err() && observed_upstream.transport_failed { + return Err(miette::Report::new(UpstreamHttpError { + response_started: observed_upstream.response_started, + })); + } + result +} + +async fn relay_http_exchange( + req: &L7Request, + client: &mut C, + upstream: &mut U, + options: RelayRequestOptions<'_>, + response_middleware: Option>, + observer: Option<&EndpointObserver>, +) -> Result +where + C: AsyncRead + AsyncWrite + Unpin, + U: AsyncRead + AsyncWrite + Unpin, +{ ensure_credential_generation_current(options)?; let header_end = req .raw_header @@ -1296,6 +1346,18 @@ fn ensure_body_generation_current(options: RelayRequestOptions<'_>) -> Result<() struct ObservedUpstream<'a, U> { upstream: &'a mut U, observer: Option<&'a EndpointObserver>, + response_started: bool, + transport_failed: bool, +} + +impl ObservedUpstream<'_, U> { + fn transport_error(&mut self, error: std::io::Error) -> std::io::Error { + self.transport_failed = true; + if let Some(observer) = self.observer { + observer.observe(EndpointResult::TransportFailed); + } + error + } } impl AsyncRead for ObservedUpstream<'_, U> { @@ -1305,13 +1367,15 @@ impl AsyncRead for ObservedUpstream<'_, U> { buf: &mut tokio::io::ReadBuf<'_>, ) -> std::task::Poll> { let this = self.get_mut(); + let before = buf.filled().len(); let result = std::pin::Pin::new(&mut *this.upstream).poll_read(cx, buf); - if matches!(result, std::task::Poll::Ready(Err(_))) - && let Some(observer) = this.observer - { - observer.observe(EndpointResult::TransportFailed); + this.response_started |= buf.filled().len() > before; + match result { + std::task::Poll::Ready(Err(error)) => { + std::task::Poll::Ready(Err(this.transport_error(error))) + } + result => result, } - result } } @@ -1323,12 +1387,15 @@ impl AsyncWrite for ObservedUpstream<'_, U> { ) -> std::task::Poll> { let this = self.get_mut(); let result = std::pin::Pin::new(&mut *this.upstream).poll_write(cx, buf); - if matches!(result, std::task::Poll::Ready(Err(_))) - && let Some(observer) = this.observer - { - observer.observe(EndpointResult::TransportFailed); + match result { + std::task::Poll::Ready(Err(error)) => { + std::task::Poll::Ready(Err(this.transport_error(error))) + } + std::task::Poll::Ready(Ok(0)) if !buf.is_empty() => std::task::Poll::Ready(Err( + this.transport_error(std::io::Error::from(std::io::ErrorKind::WriteZero)) + )), + result => result, } - result } fn poll_flush( @@ -1337,12 +1404,12 @@ impl AsyncWrite for ObservedUpstream<'_, U> { ) -> std::task::Poll> { let this = self.get_mut(); let result = std::pin::Pin::new(&mut *this.upstream).poll_flush(cx); - if matches!(result, std::task::Poll::Ready(Err(_))) - && let Some(observer) = this.observer - { - observer.observe(EndpointResult::TransportFailed); + match result { + std::task::Poll::Ready(Err(error)) => { + std::task::Poll::Ready(Err(this.transport_error(error))) + } + result => result, } - result } fn poll_shutdown( @@ -6288,7 +6355,10 @@ mod tests { } } - async fn assert_observed_response_result(response: &'static [u8], expected: EndpointResult) { + async fn assert_observed_response_result( + response: &'static [u8], + expected: EndpointResult, + ) -> Result { let (observer, mut receiver) = test_endpoint_observer().await; let (mut upstream, mut upstream_peer) = tokio::io::duplex(4096); let writer = tokio::spawn(async move { @@ -6298,7 +6368,7 @@ mod tests { .expect("write response"); }); let mut client = tokio::io::sink(); - relay_response( + let result = relay_response( "POST", &mut upstream, &mut client, @@ -6308,14 +6378,14 @@ mod tests { }, None, ) - .await - .expect("relay observed response"); + .await; writer.await.expect("join upstream writer"); let command = receiver.recv().await.expect("endpoint observation"); assert!(matches!( command, EndpointStatusCommand::Observe { result, .. } if result == expected )); + result } #[tokio::test] @@ -6324,7 +6394,8 @@ mod tests { b"HTTP/1.1 204 No Content\r\n\r\n", EndpointResult::HttpResponseReceived, ) - .await; + .await + .unwrap(); } #[tokio::test] @@ -6333,7 +6404,9 @@ mod tests { &b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n"[..], &b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n"[..], ] { - assert_observed_response_result(response, EndpointResult::UpstreamRejected).await; + assert_observed_response_result(response, EndpointResult::UpstreamRejected) + .await + .unwrap(); } } @@ -6412,7 +6485,15 @@ mod tests { &b"HTTP/1.1 100 Continue\r\n\r\n"[..], &b"HTTP/1.1 103 Early Hints\r\n\r\n"[..], ] { - assert_observed_response_result(response, EndpointResult::TransportFailed).await; + let error = assert_observed_response_result(response, EndpointResult::TransportFailed) + .await + .expect_err("interim response without a final response"); + assert!( + error + .downcast_ref::() + .unwrap() + .response_started + ); } } @@ -6481,6 +6562,8 @@ mod tests { let mut upstream = ObservedUpstream { upstream: &mut response, observer: Some(&observer), + response_started: false, + transport_failed: false, }; let mut client = Vec::new(); relay_response( @@ -6513,7 +6596,8 @@ mod tests { b"not-http 200 nope\r\nContent-Length: 0\r\n\r\n", EndpointResult::TransportFailed, ) - .await; + .await + .unwrap(); } #[tokio::test] diff --git a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs index 3efc323bdb..a0ef69c371 100644 --- a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs +++ b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs @@ -98,7 +98,9 @@ where if !buf.is_empty() { client.write_all(&buf).await.into_diagnostic()?; } - return Ok(RelayOutcome::Consumed); + return Err(miette::Report::new(UpstreamHttpError { + response_started: !buf.is_empty() || informational_header_bytes > 0, + })); } buf.extend_from_slice(&tmp[..n]); }; diff --git a/docs/how-it-works/policies/network-rules.mdx b/docs/how-it-works/policies/network-rules.mdx index 66405e5e5b..98426f2374 100644 --- a/docs/how-it-works/policies/network-rules.mdx +++ b/docs/how-it-works/policies/network-rules.mdx @@ -115,6 +115,13 @@ it. For a protocol in which the server sends first, such as SMTP, set Use an inspected endpoint when the request, not only the destination, determines the risk, such as allowing reads but not writes on an API. +For HTTP traffic that OpenShell parses inside a CONNECT tunnel, it closes the +client connection when the upstream closes while idle. If forwarding fails before the +upstream sends any response bytes, OpenShell returns `502 Bad Gateway` with +`Connection: close` and logs the upstream failure. OpenShell does not replay +the request: a missing response does not prove that the upstream did not execute +it. Once a response has started, a transport failure closes the connection. + ### Enforcement The `enforcement` field on an inspected endpoint decides what happens when a From e5fe0357efcce7a15ef428312cb15051185d359c Mon Sep 17 00:00:00 2001 From: Shiju Date: Fri, 6 Nov 2026 20:34:36 +0530 Subject: [PATCH 2/4] test(proxy): tighten upstream close coverage Consolidate overlapping forwarding cases through the complete relay and retain distinct failure boundaries. Remove relay implementation details from the network policy guide. Signed-off-by: Shiju --- .../src/l7/relay/upstream_close_tests.rs | 143 ++++++------------ docs/how-it-works/policies/network-rules.mdx | 7 - 2 files changed, 46 insertions(+), 104 deletions(-) diff --git a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs index 8924b717fb..efd9983809 100644 --- a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs +++ b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs @@ -2,9 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use super::*; -use crate::l7::provider::{BodyLength, L7Request}; use crate::opa::{NetworkInput, OpaEngine}; -use std::collections::HashMap; use std::path::PathBuf; use std::time::Duration; use tokio::net::{TcpListener, TcpStream}; @@ -129,56 +127,41 @@ network_policies: } #[tokio::test] -async fn queued_request_after_upstream_close_returns_502() { - timeout(Duration::from_secs(5), async { - let engine = OpaEngine::from_strings( - include_str!("../../../data/sandbox-policy.rego"), - "network_policies: {}\n", - ) - .unwrap(); - let (mut app, mut client) = tokio::io::duplex(8192); - let (mut relay_upstream, mut upstream) = tokio::io::duplex(8192); - // Both requests are buffered before the relay starts. When the first - // response arrives, request two and the upstream close are ready together. - app.write_all(&[REQUEST, REQUEST].concat()).await.unwrap(); - let relay = tokio::spawn(async move { - let guard = engine - .generation_guard(engine.current_generation()) - .unwrap(); - relay_passthrough_with_credentials( - &mut client, - &mut relay_upstream, - &context(), - &guard, - Some(&engine), - ) - .await - }); - read_exact_message(&mut upstream, REQUEST).await; - upstream.write_all(RESPONSE).await.unwrap(); - drop(upstream); - let mut received = Vec::new(); - app.read_to_end(&mut received).await.unwrap(); - assert_eq!(received, [RESPONSE, BAD_GATEWAY].concat()); - relay.await.unwrap().unwrap(); - }) - .await - .expect("queued request must receive an explicit gateway error"); -} - -#[tokio::test] -async fn upstream_failure_returns_502_only_before_response_starts() { +async fn queued_request_handles_upstream_close_without_replay() { timeout(Duration::from_secs(5), async { + // None closes before request two is forwarded (BrokenPipe). The other + // cases accept it, then close before or during the response headers. for response in [ - None, // Upstream disappeared before the first write. - Some(&b""[..]), // Request was accepted, then upstream closed without responding. + None, + Some(&b""[..]), Some(&b"HTTP/1.1 100 Continue\r\n\r\n"[..]), Some(&b"HTTP/1.1 200"[..]), - Some(&b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nOK"[..]), - Some(&b"HTTP/1.0 200 OK\r\n\r\nOK"[..]), // Valid close-delimited response. ] { + let engine = OpaEngine::from_strings( + include_str!("../../../data/sandbox-policy.rego"), + "network_policies: {}\n", + ) + .unwrap(); let (mut app, mut client) = tokio::io::duplex(8192); let (mut relay_upstream, mut upstream) = tokio::io::duplex(8192); + // Buffer both requests so request two and the upstream close can + // become ready together after the first successful exchange. + app.write_all(&[REQUEST, REQUEST].concat()).await.unwrap(); + let relay = tokio::spawn(async move { + let guard = engine + .generation_guard(engine.current_generation()) + .unwrap(); + relay_passthrough_with_credentials( + &mut client, + &mut relay_upstream, + &context(), + &guard, + Some(&engine), + ) + .await + }); + read_exact_message(&mut upstream, REQUEST).await; + upstream.write_all(RESPONSE).await.unwrap(); let server = response.map(|response| { tokio::spawn(async move { read_exact_message(&mut upstream, REQUEST).await; @@ -189,80 +172,46 @@ async fn upstream_failure_returns_502_only_before_response_starts() { assert!(extra.is_empty(), "POST must not be replayed"); }) }); - let relay = tokio::spawn(async move { - let req = L7Request { - action: "POST".into(), - target: "/mcp".into(), - query_params: HashMap::new(), - raw_header: REQUEST.to_vec(), - body_length: BodyLength::ContentLength(2), - }; - relay_http_request_with_credential_rejection( - &req, - &mut client, - &mut relay_upstream, - crate::l7::rest::RelayRequestOptions::default(), - &context(), - None, - ) - .await - }); let mut received = Vec::new(); app.read_to_end(&mut received).await.unwrap(); - let result = relay.await.unwrap(); - if response.is_none_or(<[u8]>::is_empty) { - assert_eq!(received, BAD_GATEWAY); - assert!(result.unwrap().is_none()); - } else { - let expected = if response == Some(&b"HTTP/1.0 200 OK\r\n\r\nOK"[..]) { - &b"HTTP/1.0 200 OK\r\nConnection: close\r\n\r\nOK"[..] - } else { - response.unwrap() - }; - assert_eq!(received, expected); - assert!( - !received - .windows(15) - .any(|bytes| bytes == b"502 Bad Gateway") - ); - } + let expected = response + .filter(|bytes| !bytes.is_empty()) + .unwrap_or(BAD_GATEWAY); + assert_eq!(received, [RESPONSE, expected].concat()); + relay.await.unwrap().unwrap(); if let Some(server) = server { server.await.unwrap(); } } }) .await - .expect("failed upstream exchange must close promptly"); + .expect("queued request must finish with a response or a clean close"); } #[tokio::test] async fn client_body_eof_is_not_an_upstream_failure() { timeout(Duration::from_secs(5), async { + let engine = OpaEngine::from_strings( + include_str!("../../../data/sandbox-policy.rego"), + "network_policies: {}\n", + ) + .unwrap(); + let guard = engine + .generation_guard(engine.current_generation()) + .unwrap(); let (mut app, mut client) = tokio::io::duplex(8192); let (mut relay_upstream, _upstream) = tokio::io::duplex(8192); + app.write_all(&REQUEST[..REQUEST.len() - 2]).await.unwrap(); app.shutdown().await.unwrap(); - let req = L7Request { - action: "POST".into(), - target: "/mcp".into(), - query_params: HashMap::new(), - raw_header: REQUEST[..REQUEST.len() - 2].to_vec(), - body_length: BodyLength::ContentLength(2), - }; - let error = relay_http_request_with_credential_rejection( - &req, + relay_passthrough_with_credentials( &mut client, &mut relay_upstream, - crate::l7::rest::RelayRequestOptions::default(), &context(), - None, + &guard, + Some(&engine), ) .await .expect_err("client did not provide its declared body"); - assert!( - error - .downcast_ref::() - .is_none() - ); drop(client); let mut response = Vec::new(); app.read_to_end(&mut response).await.unwrap(); diff --git a/docs/how-it-works/policies/network-rules.mdx b/docs/how-it-works/policies/network-rules.mdx index 98426f2374..66405e5e5b 100644 --- a/docs/how-it-works/policies/network-rules.mdx +++ b/docs/how-it-works/policies/network-rules.mdx @@ -115,13 +115,6 @@ it. For a protocol in which the server sends first, such as SMTP, set Use an inspected endpoint when the request, not only the destination, determines the risk, such as allowing reads but not writes on an API. -For HTTP traffic that OpenShell parses inside a CONNECT tunnel, it closes the -client connection when the upstream closes while idle. If forwarding fails before the -upstream sends any response bytes, OpenShell returns `502 Bad Gateway` with -`Connection: close` and logs the upstream failure. OpenShell does not replay -the request: a missing response does not prove that the upstream did not execute -it. Once a response has started, a transport failure closes the connection. - ### Enforcement The `enforcement` field on an inspected endpoint decides what happens when a From c89327f768974e93a6c7eeb4d6f0134d26d38024 Mon Sep 17 00:00:00 2001 From: Shiju Date: Fri, 6 Nov 2026 22:47:59 +0530 Subject: [PATCH 3/4] fix(proxy): complete HTTP upstream failure handling Cover JSON-RPC/MCP idle closure and forward HTTP failures. Track response commitment at client writes, retain upstream causes, and distinguish truncated framing from normal completion and policy reload. Preserve request correlation and upstream/session cleanup. Signed-off-by: Shiju --- .../src/l7/relay.rs | 237 +++++++---- .../src/l7/relay/upstream_close_tests.rs | 388 ++++++++++++------ .../src/l7/rest.rs | 189 +++++++-- .../src/l7/rest/http_response.rs | 20 +- .../openshell-supervisor-network/src/proxy.rs | 135 +++++- 5 files changed, 707 insertions(+), 262 deletions(-) diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 671f739e3c..19a7d1b432 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -20,6 +20,7 @@ use crate::l7::provider::{L7Provider, RelayOutcome}; use crate::l7::rest::WebSocketExtensionMode; use crate::l7::{EndpointObserver, EnforcementMode, L7EndpointConfig, L7Protocol, L7RequestInfo}; use crate::opa::{PolicyGenerationGuard, TunnelPolicyEngine}; +use futures::FutureExt; use miette::{IntoDiagnostic, Result, miette}; use openshell_core::activity::{ActivitySender, try_record_activity}; use openshell_core::endpoint_status::{EndpointObservationSender, EndpointResult}; @@ -755,6 +756,7 @@ where client, upstream, crate::l7::rest::RelayRequestOptions { + request_id: Some(request_id), resolver: ctx.secret_resolver.as_deref(), body_classifier: ctx.body_classifier.as_deref(), mcp_request_validation: (config.protocol == L7Protocol::Mcp).then_some( @@ -819,7 +821,7 @@ where fn build_upstream_http_failure_event( ctx: &L7EvalContext, method: &str, - response_started: bool, + response_code: Option, ) -> openshell_ocsf::OcsfEvent { let mut event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::for_http_method(method)) @@ -831,12 +833,42 @@ fn build_upstream_http_failure_event( .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) .message("HTTP upstream connection closed or failed") .status_detail("upstream_http_transport_failure"); - if !response_started { - event = event.http_response(openshell_ocsf::HttpResponse { code: 502 }); + if let Some(code) = response_code { + event = event.http_response(openshell_ocsf::HttpResponse { code }); } event.build() } +pub(crate) async fn handle_upstream_http_error( + client: &mut C, + upstream: &mut U, + ctx: &L7EvalContext, + method: &str, + request_id: Option<&str>, + error: &crate::l7::rest::UpstreamHttpError, +) -> Result<()> +where + C: AsyncWrite + Unpin, + U: AsyncWrite + Unpin, +{ + debug!(error = ?error, "HTTP upstream transport failure"); + let _ = upstream.shutdown().await; + // An unanswered request may already have executed. Never replay it or + // append a new status after the client stream accepted response bytes. + let delivery = if error.response_started { + Ok(()) + } else { + write_bad_gateway_response(client, request_id).await + }; + let response_code = (!error.response_started && delivery.is_ok()).then_some(502); + ocsf_emit!(build_upstream_http_failure_event( + ctx, + method, + response_code + )); + delivery +} + async fn relay_http_request_with_credential_rejection_observed( request: &crate::l7::provider::L7Request, client: &mut C, @@ -865,17 +897,16 @@ where Ok(outcome) => Ok(Some(outcome)), Err(report) => { if let Some(error) = report.downcast_ref::() { - ocsf_emit!(build_upstream_http_failure_event( + handle_upstream_http_error( + client, + upstream, ctx, &request.action, - error.response_started, - )); - // A missing response does not prove the request was not executed. - // Never replay it, and never append a 502 to a started response. - if !error.response_started { - write_bad_gateway_response(client, None).await?; - } - Ok(None) + options.request_id, + error, + ) + .await?; + Ok(Some(RelayOutcome::Consumed)) } else if let Some(error) = report.downcast_ref::() { // Body classification includes credentials for other destinations, @@ -1247,25 +1278,26 @@ where .endpoint_observation_tx .as_ref() .and_then(EndpointObservationSender::capture); - let mut req = match parse_next_http_request(&provider, client, upstream, ctx).await { - Ok(Some(req)) => req, - Ok(None) => return Ok(()), - Err(e) => { - if is_benign_connection_error(&e) { - debug!( - host = %ctx.host, - port = ctx.port, - error = %e, - "L7 route-selected connection closed" - ); - } else { - let detail = - parse_rejection_detail(&e.to_string(), ParseRejectionMode::L7Endpoint); - emit_parse_rejection(ctx, &detail, "l7"); + let mut req = + match parse_next_http_request(provider.parse_request(client), upstream, ctx).await { + Ok(Some(req)) => req, + Ok(None) => return Ok(()), + Err(e) => { + if is_benign_connection_error(&e) { + debug!( + host = %ctx.host, + port = ctx.port, + error = %e, + "L7 route-selected connection closed" + ); + } else { + let detail = + parse_rejection_detail(&e.to_string(), ParseRejectionMode::L7Endpoint); + emit_parse_rejection(ctx, &detail, "l7"); + } + return Ok(()); } - return Ok(()); - } - }; + }; if !request_authority_matches_endpoint(&req, ctx) { reject_request_authority_mismatch(client, ctx, &req.action).await?; return Ok(()); @@ -2113,9 +2145,10 @@ where .filter(|_| ctx.token_grant_resolver.is_some()) .and_then(EndpointObservationSender::capture); // Parse one HTTP request from client - let req = match parse_next_http_request(&provider, client, upstream, ctx).await { + let req = match parse_next_http_request(provider.parse_request(client), upstream, ctx).await + { Ok(Some(req)) => req, - Ok(None) => return Ok(()), // Client closed connection + Ok(None) => return Ok(()), // The HTTP connection is no longer reusable. Err(e) => { if is_benign_connection_error(&e) { debug!( @@ -2523,14 +2556,18 @@ where .endpoint_observation_tx .as_ref() .and_then(EndpointObservationSender::capture); - let parsed = match crate::l7::jsonrpc::parse_jsonrpc_http_request( - client, - config.json_rpc_max_body_bytes, - crate::l7::path::CanonicalizeOptions { - allow_encoded_slash: config.allow_encoded_slash, - ..Default::default() - }, - crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(config), + let parsed = match parse_next_http_request( + Box::pin(crate::l7::jsonrpc::parse_jsonrpc_http_request( + client, + config.json_rpc_max_body_bytes, + crate::l7::path::CanonicalizeOptions { + allow_encoded_slash: config.allow_encoded_slash, + ..Default::default() + }, + crate::l7::jsonrpc::JsonRpcInspectionOptions::for_config(config), + )), + upstream, + ctx, ) .await { @@ -2810,25 +2847,26 @@ where // Validate the head, including body framing, before deciding whether // this endpoint can inspect the requested protocol. Upgrade refusal // must not wait for a body or depend on its inspection size limit. - let mut req = match parse_next_http_request(&provider, client, upstream, ctx).await { - Ok(Some(req)) => req, - Ok(None) => return Ok(()), - Err(e) => { - if is_benign_connection_error(&e) { - debug!( - host = %ctx.host, - port = ctx.port, - error = %e, - "GraphQL L7 connection closed" - ); - } else { - let detail = - parse_rejection_detail(&e.to_string(), ParseRejectionMode::L7Endpoint); - emit_parse_rejection(ctx, &detail, "l7-graphql"); + let mut req = + match parse_next_http_request(provider.parse_request(client), upstream, ctx).await { + Ok(Some(req)) => req, + Ok(None) => return Ok(()), + Err(e) => { + if is_benign_connection_error(&e) { + debug!( + host = %ctx.host, + port = ctx.port, + error = %e, + "GraphQL L7 connection closed" + ); + } else { + let detail = + parse_rejection_detail(&e.to_string(), ParseRejectionMode::L7Endpoint); + emit_parse_rejection(ctx, &detail, "l7-graphql"); + } + return Ok(()); } - return Ok(()); - } - }; + }; if !request_authority_matches_endpoint(&req, ctx) { reject_request_authority_mismatch(client, ctx, &req.action).await?; @@ -3570,39 +3608,70 @@ fn evaluate_l7_request_once( /// both sides are ready, finish the request first so a forwarding failure can /// produce an explicit HTTP error. Unsolicited upstream bytes are not a response /// to a request and cannot be assigned to the next exchange. -async fn parse_next_http_request( - provider: &crate::l7::rest::RestProvider, - client: &mut C, +async fn parse_next_http_request( + request: F, upstream: &mut U, ctx: &L7EvalContext, -) -> Result> +) -> Result> where - C: AsyncRead + AsyncWrite + Unpin + Send, + F: Future>>, U: AsyncRead + Unpin, { let mut byte = [0u8; 1]; - tokio::select! { + let request = tokio::select! { biased; - request = provider.parse_request(client) => request, + request = request => request, result = upstream.read(&mut byte) => { - let (detail, status, severity) = match result { - Ok(0) => ("HTTP upstream closed while idle", StatusId::Success, SeverityId::Informational), - Ok(_) => ("HTTP upstream sent unsolicited data while idle", StatusId::Failure, SeverityId::Low), - Err(_) => ("HTTP upstream failed while idle", StatusId::Failure, SeverityId::Low), - }; - let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) - .activity(ActivityId::Close) - .action(ActionId::Allowed) - .disposition(DispositionId::Allowed) - .severity(severity) - .status(status) - .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) - .message(detail) - .build(); - ocsf_emit!(event); - Ok(None) + emit_idle_upstream_close(ctx, result); + return Ok(None); + } + }; + // Parsing wins a simultaneous close so forwarding can return 502, but + // ready unsolicited data must never become that request's response. + if matches!(request, Ok(Some(_))) { + match upstream.read(&mut byte).now_or_never() { + Some(Ok(n)) if n > 0 => { + emit_idle_upstream_close(ctx, Ok(n)); + return Ok(None); + } + Some(Err(error)) => { + debug!(host = %ctx.host, port = ctx.port, error = %error, + "HTTP upstream failed as a client request became ready"); + } + _ => {} } } + request +} + +fn emit_idle_upstream_close(ctx: &L7EvalContext, result: std::io::Result) { + let (detail, status, severity) = match result { + Ok(0) => ( + "HTTP upstream closed while idle", + StatusId::Success, + SeverityId::Informational, + ), + Ok(_) => ( + "HTTP upstream sent unsolicited data while idle", + StatusId::Failure, + SeverityId::Low, + ), + Err(_) => ( + "HTTP upstream failed while idle", + StatusId::Failure, + SeverityId::Low, + ), + }; + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Close) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(severity) + .status(status) + .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) + .message(detail) + .build(); + ocsf_emit!(event); } /// Relay HTTP traffic with credential injection only (no L7 OPA evaluation). @@ -3639,9 +3708,10 @@ where .filter(|_| ctx.token_grant_resolver.is_some()) .and_then(EndpointObservationSender::capture); // Read next request from client. - let req = match parse_next_http_request(&provider, client, upstream, ctx).await { + let req = match parse_next_http_request(provider.parse_request(client), upstream, ctx).await + { Ok(Some(req)) => req, - Ok(None) => break, // Client closed connection. + Ok(None) => break, // The HTTP connection is no longer reusable. Err(e) => { if is_benign_connection_error(&e) { break; @@ -3799,6 +3869,7 @@ where client, upstream, crate::l7::rest::RelayRequestOptions { + request_id: Some(&request_id), resolver, credential_generation: credential_generation_guard(ctx), generation_guard: Some(generation_guard), diff --git a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs index efd9983809..8e9de1e871 100644 --- a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs +++ b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs @@ -11,8 +11,6 @@ use tokio::time::timeout; const REQUEST: &[u8] = b"POST /mcp HTTP/1.1\r\nHost: api.example.test\r\nContent-Length: 2\r\n\r\n{}"; const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"; -const BAD_GATEWAY: &[u8] = - b"HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; fn context() -> L7EvalContext { L7EvalContext { @@ -42,88 +40,105 @@ async fn tcp_pair() -> (TcpStream, TcpStream) { (client, server) } +async fn relay_fixture(protocol: &str, mut client: C, mut upstream: U) -> Result<()> +where + C: AsyncRead + AsyncWrite + Unpin + Send, + U: AsyncRead + AsyncWrite + Unpin + Send, +{ + let ctx = context(); + let policy_protocol = if protocol == "passthrough" { + "rest" + } else { + protocol + }; + let rule = if policy_protocol == "rest" { + serde_json::json!({"method":"POST", "path":"/mcp"}) + } else { + serde_json::json!({"method":"initialize"}) + }; + let data = serde_json::json!({"network_policies": {"rest_api": { + "name":"rest_api", "endpoints":[{"host":ctx.host, "port":ctx.port, + "path":"/mcp", "protocol":policy_protocol, "enforcement":"enforce", "rules":[{"allow":rule}]}], + "binaries":[{"path":ctx.binary_path}] + }}}); + let engine = OpaEngine::from_strings( + include_str!("../../../data/sandbox-policy.rego"), + &data.to_string(), + ) + .unwrap(); + if protocol == "passthrough" { + let guard = engine + .generation_guard(engine.current_generation()) + .unwrap(); + relay_passthrough_with_credentials(&mut client, &mut upstream, &ctx, &guard, Some(&engine)) + .await + } else { + let input = NetworkInput { + host: ctx.host.clone(), + port: ctx.port, + binary_path: PathBuf::from(&ctx.binary_path), + binary_sha256: "unused".into(), + ancestors: vec![], + cmdline_paths: vec![], + }; + let (config, generation) = engine + .query_endpoint_config_with_generation(&input) + .unwrap(); + let config = crate::l7::parse_l7_config(&config.unwrap()).unwrap(); + relay_with_inspection( + &config, + engine.clone_engine_for_tunnel(generation).unwrap(), + &mut client, + &mut upstream, + &ctx, + ) + .await + } +} + +fn assert_bad_gateway(bytes: &[u8]) { + let response = std::str::from_utf8(bytes).unwrap(); + assert!( + response.starts_with("HTTP/1.1 502 Bad Gateway\r\n"), + "{response:?}" + ); + assert!(response.contains("Content-Length: 0\r\n")); + assert!(response.ends_with("Connection: close\r\n\r\n")); + let prefix = format!( + "{}: ", + openshell_core::token_grant_failure::REQUEST_ID_HEADER + ); + let id = response + .lines() + .find_map(|line| line.strip_prefix(&prefix)) + .expect("correlation header"); + uuid::Uuid::parse_str(id).unwrap(); +} + #[tokio::test] async fn keep_alive_propagates_upstream_idle_close_over_tcp() { timeout(Duration::from_secs(5), async { - for inspected in [false, true] { - let engine = OpaEngine::from_strings( - include_str!("../../../data/sandbox-policy.rego"), - r" -network_policies: - rest_api: - name: rest_api - endpoints: - - host: api.example.test - port: 80 - protocol: rest - enforcement: enforce - rules: - - allow: { method: POST, path: /mcp } - binaries: - - { path: /usr/bin/curl } -", - ) - .unwrap(); - let (mut app, mut client) = tcp_pair().await; - let (mut relay_upstream, mut upstream) = tcp_pair().await; - let relay = tokio::spawn(async move { - let ctx = context(); - if inspected { - let input = NetworkInput { - host: ctx.host.clone(), - port: ctx.port, - binary_path: PathBuf::from(&ctx.binary_path), - binary_sha256: "unused".into(), - ancestors: vec![], - cmdline_paths: vec![], - }; - let (config, generation) = engine - .query_endpoint_config_with_generation(&input) - .unwrap(); - let config = crate::l7::parse_l7_config(&config.unwrap()).unwrap(); - let evaluator = engine.clone_engine_for_tunnel(generation).unwrap(); - relay_with_inspection( - &config, - evaluator, - &mut client, - &mut relay_upstream, - &ctx, - ) - .await - } else { - let guard = engine - .generation_guard(engine.current_generation()) - .unwrap(); - relay_passthrough_with_credentials( - &mut client, - &mut relay_upstream, - &ctx, - &guard, - Some(&engine), - ) - .await - } - }); - // Healthy keep-alive must still carry more than one complete exchange. + for protocol in ["passthrough", "rest", "json-rpc", "mcp"] { + let body = if matches!(protocol, "json-rpc" | "mcp") { + r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"test","version":"1"}}}"# + } else { "{}" }; + let request = format!("POST /mcp HTTP/1.1\r\nHost: api.example.test\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len()); + let (mut app, client) = tcp_pair().await; + let (relay_upstream, mut upstream) = tcp_pair().await; + let relay = tokio::spawn(relay_fixture(protocol, client, relay_upstream)); for _ in 0..2 { - app.write_all(REQUEST).await.unwrap(); - read_exact_message(&mut upstream, REQUEST).await; + app.write_all(request.as_bytes()).await.unwrap(); + read_exact_message(&mut upstream, request.as_bytes()).await; upstream.write_all(RESPONSE).await.unwrap(); read_exact_message(&mut app, RESPONSE).await; } upstream.shutdown().await.unwrap(); - // Do not send another request: FIN must propagate while the client is idle. let mut rest = Vec::new(); app.read_to_end(&mut rest).await.unwrap(); - assert!( - rest.is_empty(), - "idle close must not invent an HTTP response" - ); + assert!(rest.is_empty(), "idle close must not invent an HTTP response"); relay.await.unwrap().unwrap(); } - }) - .await - .expect("idle upstream close must reach the client promptly"); + }).await.expect("idle upstream close must reach the client promptly"); } #[tokio::test] @@ -131,56 +146,51 @@ async fn queued_request_handles_upstream_close_without_replay() { timeout(Duration::from_secs(5), async { // None closes before request two is forwarded (BrokenPipe). The other // cases accept it, then close before or during the response headers. - for response in [ - None, - Some(&b""[..]), - Some(&b"HTTP/1.1 100 Continue\r\n\r\n"[..]), - Some(&b"HTTP/1.1 200"[..]), - ] { - let engine = OpaEngine::from_strings( - include_str!("../../../data/sandbox-policy.rego"), - "network_policies: {}\n", - ) - .unwrap(); - let (mut app, mut client) = tokio::io::duplex(8192); - let (mut relay_upstream, mut upstream) = tokio::io::duplex(8192); - // Buffer both requests so request two and the upstream close can - // become ready together after the first successful exchange. - app.write_all(&[REQUEST, REQUEST].concat()).await.unwrap(); - let relay = tokio::spawn(async move { - let guard = engine - .generation_guard(engine.current_generation()) - .unwrap(); - relay_passthrough_with_credentials( - &mut client, - &mut relay_upstream, - &context(), - &guard, - Some(&engine), - ) - .await - }); - read_exact_message(&mut upstream, REQUEST).await; - upstream.write_all(RESPONSE).await.unwrap(); - let server = response.map(|response| { - tokio::spawn(async move { - read_exact_message(&mut upstream, REQUEST).await; - upstream.write_all(response).await.unwrap(); - upstream.shutdown().await.unwrap(); - let mut extra = Vec::new(); - upstream.read_to_end(&mut extra).await.unwrap(); - assert!(extra.is_empty(), "POST must not be replayed"); - }) - }); - let mut received = Vec::new(); - app.read_to_end(&mut received).await.unwrap(); - let expected = response - .filter(|bytes| !bytes.is_empty()) - .unwrap_or(BAD_GATEWAY); - assert_eq!(received, [RESPONSE, expected].concat()); - relay.await.unwrap().unwrap(); - if let Some(server) = server { - server.await.unwrap(); + for protocol in ["passthrough", "rest"] { + for (response, expect_gateway) in [ + (None, true), + (Some(&b""[..]), true), + (Some(&b"HTTP/1.1 100 Continue\r\n\r\n"[..]), false), + (Some(&b"HTTP/1.1 200"[..]), true), + ( + Some(&b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nOK"[..]), + false, + ), + ( + Some(&b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nOK\r\n"[..]), + false, + ), + ] { + let (mut app, client) = tokio::io::duplex(8192); + let (relay_upstream, mut upstream) = tokio::io::duplex(8192); + // Buffer both requests so request two and the upstream close can + // become ready together after the first successful exchange. + app.write_all(&[REQUEST, REQUEST].concat()).await.unwrap(); + let relay = tokio::spawn(relay_fixture(protocol, client, relay_upstream)); + read_exact_message(&mut upstream, REQUEST).await; + upstream.write_all(RESPONSE).await.unwrap(); + let server = response.map(|response| { + tokio::spawn(async move { + read_exact_message(&mut upstream, REQUEST).await; + upstream.write_all(response).await.unwrap(); + upstream.shutdown().await.unwrap(); + let mut extra = Vec::new(); + upstream.read_to_end(&mut extra).await.unwrap(); + assert!(extra.is_empty(), "POST must not be replayed"); + }) + }); + let mut received = Vec::new(); + app.read_to_end(&mut received).await.unwrap(); + assert!(received.starts_with(RESPONSE)); + if expect_gateway { + assert_bad_gateway(&received[RESPONSE.len()..]); + } else { + assert_eq!(&received[RESPONSE.len()..], response.unwrap()); + } + relay.await.unwrap().unwrap(); + if let Some(server) = server { + server.await.unwrap(); + } } } }) @@ -227,7 +237,11 @@ async fn client_body_eof_is_not_an_upstream_failure() { #[test] fn upstream_failure_event_reports_502_without_request_secrets() { for response_started in [false, true] { - let event = build_upstream_http_failure_event(&context(), "POST", response_started); + let event = build_upstream_http_failure_event( + &context(), + "POST", + (!response_started).then_some(502), + ); let json = event.to_json().unwrap(); assert_eq!(json["class_uid"], 4002); assert_eq!(json["status_detail"], "upstream_http_transport_failure"); @@ -241,3 +255,131 @@ fn upstream_failure_event_reports_502_without_request_secrets() { openshell_ocsf_validate::validate_event(&json).unwrap(); } } + +// A reset after a scripted prefix covers the error path separately from FIN. +struct ResetAfterPrefix(&'static [u8], bool); +impl AsyncRead for ResetAfterPrefix { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + if this.0.is_empty() { + std::task::Poll::Ready(Err(std::io::ErrorKind::ConnectionReset.into())) + } else { + let n = this.0.len().min(buf.remaining()); + buf.put_slice(&this.0[..n]); + this.0 = &this.0[n..]; + std::task::Poll::Ready(Ok(())) + } + } +} +impl AsyncWrite for ResetAfterPrefix { + fn poll_write( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + bytes: &[u8], + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(bytes.len())) + } + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.get_mut().1 = true; + std::task::Poll::Ready(Ok(())) + } +} + +#[tokio::test] +async fn upstream_reset_uses_client_commitment_and_preserves_cause() { + timeout(Duration::from_secs(5), async { + for (prefix, committed) in [ + (&b""[..], false), + (&b"HTTP/1.1 200"[..], false), + (&b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nOK"[..], true), + ] { + let (mut app, mut client) = tokio::io::duplex(8192); + let mut upstream = ResetAfterPrefix(prefix, false); + let request = crate::l7::rest::request_from_buffered_http( + "POST", + "/mcp", + "/mcp", + REQUEST.to_vec(), + ) + .unwrap(); + let report = + crate::l7::rest::relay_http_request_with_response_middleware_guarded_observed( + &request, + &mut client, + &mut upstream, + crate::l7::rest::RelayRequestOptions::default(), + None, + None, + ) + .await + .expect_err("scripted upstream reset"); + assert!( + report + .chain() + .any(|cause| cause.to_string().to_lowercase().contains("reset")), + "original I/O cause must remain available: {report:?}" + ); + let error = report + .downcast_ref::() + .unwrap(); + assert_eq!(error.response_started, committed); + let request_id = "9b8a7f6e-5d4c-4b3a-9f8e-7d6c5b4a3f2e"; + handle_upstream_http_error( + &mut client, + &mut upstream, + &context(), + "POST", + Some(request_id), + error, + ) + .await + .unwrap(); + assert!(upstream.1, "failed upstream must be shut down"); + drop(client); + let mut response = Vec::new(); + app.read_to_end(&mut response).await.unwrap(); + if committed { + assert_eq!(response, prefix); + } else { + assert_bad_gateway(&response); + assert!(std::str::from_utf8(&response).unwrap().contains(request_id)); + } + } + }) + .await + .expect("reset must terminate the exchange"); +} + +#[tokio::test] +async fn ready_request_does_not_consume_unsolicited_upstream_response() { + for unsolicited in [false, true] { + let (mut upstream, mut server) = tokio::io::duplex(1024); + if unsolicited { + server.write_all(RESPONSE).await.unwrap(); + } else { + server.shutdown().await.unwrap(); + } + let result = + parse_next_http_request(std::future::ready(Ok(Some(()))), &mut upstream, &context()) + .await + .unwrap(); + assert_eq!( + result.is_none(), + unsolicited, + "EOF must retain the request for the 502 path; unsolicited data must close" + ); + } +} diff --git a/crates/openshell-supervisor-network/src/l7/rest.rs b/crates/openshell-supervisor-network/src/l7/rest.rs index 0013063067..73a0ed0a46 100644 --- a/crates/openshell-supervisor-network/src/l7/rest.rs +++ b/crates/openshell-supervisor-network/src/l7/rest.rs @@ -747,6 +747,7 @@ where client, upstream, RelayRequestOptions { + request_id: None, resolver, body_classifier: None, mcp_request_validation: None, @@ -774,6 +775,7 @@ pub(crate) enum WebSocketExtensionMode { #[derive(Clone, Copy, Default)] pub(crate) struct RelayRequestOptions<'a> { + pub(crate) request_id: Option<&'a str>, pub(crate) resolver: Option<&'a SecretResolver>, pub(crate) body_classifier: Option<&'a openshell_core::secrets::body::BodyCredentialClassifier>, /// Revalidate buffered MCP requests after header transformations. @@ -830,10 +832,18 @@ impl std::error::Error for CredentialUnavailableError {} impl miette::Diagnostic for CredentialUnavailableError {} +/// A body ended before its declared framing was complete. The upstream +/// adapter identifies the origin when the same readers handle request bodies. +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +struct IncompleteHttpBody(String); +impl miette::Diagnostic for IncompleteHttpBody {} + /// An upstream transport failure, distinct from client I/O and credential refusals. #[derive(Debug)] pub(crate) struct UpstreamHttpError { pub(crate) response_started: bool, + source: miette::Report, } impl fmt::Display for UpstreamHttpError { @@ -842,7 +852,11 @@ impl fmt::Display for UpstreamHttpError { } } -impl std::error::Error for UpstreamHttpError {} +impl std::error::Error for UpstreamHttpError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&*self.source) + } +} impl miette::Diagnostic for UpstreamHttpError {} impl<'a> CredentialGenerationGuard<'a> { @@ -919,26 +933,43 @@ where let mut observed_upstream = ObservedUpstream { upstream, observer, - response_started: false, transport_failed: false, + read_eof: false, + }; + let mut client = ResponseProgress { + client, + response_started: false, }; let result = relay_http_exchange( req, - client, + &mut client, &mut observed_upstream, options, response_middleware, observer, ) .await; - // IntoDiagnostic hides the concrete I/O error. Classify at the adapter's - // owner, where an upstream failure remains distinct from client I/O. - if result.is_err() && observed_upstream.transport_failed { - return Err(miette::Report::new(UpstreamHttpError { - response_started: observed_upstream.response_started, - })); + // Classify at the I/O adapters' owner: only bytes accepted by the + // client stream commit a response, and clean body EOF is not a failure. + match result { + Err(mut report) => { + if let Some(error) = report.downcast_mut::() { + error.response_started = client.response_started; + Err(report) + } else if observed_upstream.transport_failed + || (observed_upstream.read_eof + && report.downcast_ref::().is_some()) + { + Err(miette::Report::new(UpstreamHttpError { + response_started: client.response_started, + source: report, + })) + } else { + Err(report) + } + } + result => result, } - result } async fn relay_http_exchange( @@ -1339,6 +1370,52 @@ fn ensure_body_generation_current(options: RelayRequestOptions<'_>) -> Result<() Ok(()) } +/// Track bytes accepted by the client independently of upstream buffering. +struct ResponseProgress<'a, C> { + client: &'a mut C, + response_started: bool, +} + +impl AsyncRead for ResponseProgress<'_, C> { + fn poll_read( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + std::pin::Pin::new(&mut *this.client).poll_read(cx, buf) + } +} + +impl AsyncWrite for ResponseProgress<'_, C> { + fn poll_write( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let this = self.get_mut(); + let result = std::pin::Pin::new(&mut *this.client).poll_write(cx, buf); + if let std::task::Poll::Ready(Ok(n)) = result { + this.response_started |= n > 0; + } + result + } + fn poll_flush( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + std::pin::Pin::new(&mut *this.client).poll_flush(cx) + } + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + std::pin::Pin::new(&mut *this.client).poll_shutdown(cx) + } +} + /// Upstream-only I/O adapter that cannot accidentally classify client errors. /// /// Once a valid response status consumes the observation token, later stream @@ -1346,8 +1423,8 @@ fn ensure_body_generation_current(options: RelayRequestOptions<'_>) -> Result<() struct ObservedUpstream<'a, U> { upstream: &'a mut U, observer: Option<&'a EndpointObserver>, - response_started: bool, transport_failed: bool, + read_eof: bool, } impl ObservedUpstream<'_, U> { @@ -1368,8 +1445,11 @@ impl AsyncRead for ObservedUpstream<'_, U> { ) -> std::task::Poll> { let this = self.get_mut(); let before = buf.filled().len(); + let can_read = buf.remaining() > 0; let result = std::pin::Pin::new(&mut *this.upstream).poll_read(cx, buf); - this.response_started |= buf.filled().len() > before; + this.read_eof |= matches!(result, std::task::Poll::Ready(Ok(()))) + && can_read + && buf.filled().len() == before; match result { std::task::Poll::Ready(Err(error)) => { std::task::Poll::Ready(Err(this.transport_error(error))) @@ -3266,9 +3346,9 @@ where .min(buf.len()); let n = reader.read(&mut buf[..to_read]).await.into_diagnostic()?; if n == 0 { - return Err(miette!( + return Err(miette::Report::new(IncompleteHttpBody(format!( "Connection closed with {remaining} bytes remaining" - )); + )))); } if let Some(guard) = generation_guard { guard.ensure_current()?; @@ -3456,7 +3536,9 @@ impl<'a> ChunkedRelayInput<'a> { .into_diagnostic()?; if read == 0 { self.pending.truncate(start); - return Err(miette!(eof_message)); + return Err(miette::Report::new(IncompleteHttpBody( + eof_message.to_string(), + ))); } self.pending.truncate(start + read); if let Some(guard) = generation_guard { @@ -3480,7 +3562,9 @@ impl<'a> ChunkedRelayInput<'a> { let byte = match reader.read_u8().await { Ok(byte) => byte, Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => { - return Err(miette!(eof_message)); + return Err(miette::Report::new(IncompleteHttpBody( + eof_message.to_string(), + ))); } Err(error) => return Err(error).into_diagnostic(), }; @@ -6562,8 +6646,8 @@ mod tests { let mut upstream = ObservedUpstream { upstream: &mut response, observer: Some(&observer), - response_started: false, transport_failed: false, + read_eof: false, }; let mut client = Vec::new(); relay_response( @@ -7172,6 +7256,7 @@ mod tests { async fn run_response_relay_across_policy_reload( script: ResponseRelayScript, + close_delimited: bool, ) -> (Result, Vec) { let entered = Arc::new(tokio::sync::Notify::new()); let release = Arc::new(tokio::sync::Notify::new()); @@ -7198,14 +7283,28 @@ mod tests { .expect("initial generation guard"); let mut middleware = response_middleware_context(&runner, &chain, "GET"); middleware.generation_guard = Some(&guard); - let mut upstream = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello".as_slice(); - let mut delivered = Vec::new(); - let relay = relay_response( - "GET", + let (mut upstream, mut server) = tokio::io::duplex(8192); + let (mut app, mut client) = tokio::io::duplex(8192); + let raw = b"GET / HTTP/1.1\r\nHost: example.test\r\n\r\n"; + let request = request_from_buffered_http("GET", "/", "/", raw.to_vec()).unwrap(); + let response = if close_delimited { + &b"HTTP/1.1 200 OK\r\nConnection: close\r\n\r\nhello"[..] + } else { + &b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello"[..] + }; + let writer = tokio::spawn(async move { + let mut received = vec![0; raw.len()]; + server.read_exact(&mut received).await.unwrap(); + assert_eq!(received, raw); + server.write_all(response).await.unwrap(); + }); + let relay = relay_http_request_with_response_middleware_guarded_observed( + &request, + &mut client, &mut upstream, - &mut delivered, - RelayResponseOptions::default(), + RelayRequestOptions::default(), Some(middleware), + None, ); let reload = async { entered.notified().await; @@ -7215,28 +7314,50 @@ mod tests { release.notify_one(); }; let (outcome, ()) = tokio::join!(relay, reload); + writer.await.unwrap(); + drop(client); + let mut delivered = Vec::new(); + app.read_to_end(&mut delivered).await.unwrap(); (outcome, delivered) } #[tokio::test] async fn response_middleware_rechecks_generation_after_stream_exchange() { - let (outcome, delivered) = - run_response_relay_across_policy_reload(ResponseRelayScript::Stream).await; + for close_delimited in [false, true] { + let (outcome, delivered) = run_response_relay_across_policy_reload( + ResponseRelayScript::Stream, + close_delimited, + ) + .await; - let error = outcome.expect_err("stale stream output must not be delivered"); - assert!(error.to_string().contains("policy generation is stale")); - assert!(delivered.ends_with(b"\r\n\r\n")); - assert!(!delivered.windows(5).any(|window| window == b"HELLO")); + let error = outcome.expect_err("stale stream output must not be delivered"); + assert!( + error.downcast_ref::().is_none(), + "normal EOF then policy reload is not an upstream failure" + ); + assert!(error.to_string().contains("policy generation is stale")); + assert!(delivered.ends_with(b"\r\n\r\n")); + assert!(!delivered.windows(5).any(|window| window == b"HELLO")); + } } #[tokio::test] async fn response_middleware_rechecks_generation_after_whole_body_finish() { - let (outcome, delivered) = - run_response_relay_across_policy_reload(ResponseRelayScript::WholeBody).await; + for close_delimited in [false, true] { + let (outcome, delivered) = run_response_relay_across_policy_reload( + ResponseRelayScript::WholeBody, + close_delimited, + ) + .await; - let error = outcome.expect_err("stale whole-body output must not be delivered"); - assert!(error.to_string().contains("policy generation is stale")); - assert!(delivered.is_empty()); + let error = outcome.expect_err("stale whole-body output must not be delivered"); + assert!( + error.downcast_ref::().is_none(), + "normal EOF then policy reload is not an upstream failure" + ); + assert!(error.to_string().contains("policy generation is stale")); + assert!(delivered.is_empty()); + } } #[tokio::test] diff --git a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs index a0ef69c371..7f9ea47abe 100644 --- a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs +++ b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs @@ -95,11 +95,9 @@ where if let Some(observer) = options.observer.as_ref() { observer.observe(EndpointResult::TransportFailed); } - if !buf.is_empty() { - client.write_all(&buf).await.into_diagnostic()?; - } return Err(miette::Report::new(UpstreamHttpError { - response_started: !buf.is_empty() || informational_header_bytes > 0, + response_started: informational_header_bytes > 0, + source: miette!("HTTP upstream closed before a complete response header"), })); } buf.extend_from_slice(&tmp[..n]); @@ -1365,7 +1363,9 @@ impl<'a, R: AsyncRead + Unpin> BufferedResponseReader<'a, R> { while self.exact_buffer.len() < length { let remaining = length - self.exact_buffer.len(); let Some(data) = self.read_some(remaining).await? else { - return Err(miette!("HTTP response body ended unexpectedly")); + return Err(miette::Report::new(IncompleteHttpBody( + "HTTP response body ended unexpectedly".into(), + ))); }; self.exact_buffer.extend_from_slice(&data); } @@ -1376,7 +1376,9 @@ impl<'a, R: AsyncRead + Unpin> BufferedResponseReader<'a, R> { async fn read_line(&mut self) -> Result> { loop { let Some(byte) = self.read_some(1).await? else { - return Err(miette!("HTTP response ended before line terminator")); + return Err(miette::Report::new(IncompleteHttpBody( + "HTTP response ended before line terminator".into(), + ))); }; self.line_buffer.push(byte[0]); if self.line_buffer.len() > MAX_HEADER_BYTES { @@ -1727,7 +1729,11 @@ where } else { read.await }; - let data = result?.ok_or_else(|| miette!("HTTP response body ended unexpectedly"))?; + let data = result?.ok_or_else(|| { + miette::Report::new(IncompleteHttpBody( + "HTTP response body ended unexpectedly".into(), + )) + })?; payload.extend_from_slice(&data); if payload.len() == limit { return Ok(payload); diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 85196d9f44..88a7b4a828 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -5371,6 +5371,7 @@ where client, upstream, crate::l7::rest::RelayRequestOptions { + request_id: None, resolver: options.secret_resolver, body_classifier: options.body_classifier, mcp_request_validation: None, @@ -6835,27 +6836,45 @@ async fn handle_forward_proxy( .await; let outcome_result = match outcome_result { Err(report) => { - if let Some(error) = report.downcast_ref::() { + if let Some(error) = report.downcast_ref::() { if let Some(session) = middleware_session.take() { session - .end(openshell_core::proto::MiddlewareSessionEndReason::Cancellation) + .end(openshell_core::proto::MiddlewareSessionEndReason::UpstreamFailure) .await; } - let _ = upstream.shutdown().await; - crate::l7::relay::reject_body_credential(client, *error).await?; - return Ok(()); - } - if let Some(error) = report.downcast_ref::() { - if let Some(session) = middleware_session.take() { - session - .end(openshell_core::proto::MiddlewareSessionEndReason::Cancellation) - .await; + crate::l7::relay::handle_upstream_http_error( + client, + &mut upstream, + &l7_ctx, + method, + Some(&request_id), + error, + ) + .await?; + Ok(crate::l7::provider::RelayOutcome::Consumed) + } else { + if let Some(error) = report.downcast_ref::() { + if let Some(session) = middleware_session.take() { + session + .end(openshell_core::proto::MiddlewareSessionEndReason::Cancellation) + .await; + } + let _ = upstream.shutdown().await; + crate::l7::relay::reject_body_credential(client, *error).await?; + return Ok(()); } - crate::l7::relay::reject_credential_resolution(client, &l7_ctx, method, error) - .await?; - return Ok(()); + if let Some(error) = report.downcast_ref::() { + if let Some(session) = middleware_session.take() { + session + .end(openshell_core::proto::MiddlewareSessionEndReason::Cancellation) + .await; + } + crate::l7::relay::reject_credential_resolution(client, &l7_ctx, method, error) + .await?; + return Ok(()); + } + Err(report) } - Err(report) } outcome => outcome, }; @@ -9549,6 +9568,92 @@ network_policies: {} } } + #[tokio::test] + async fn forward_upstream_eof_returns_502_and_preserves_allow_activity() { + let Some(ip) = non_loopback_test_ipv4() else { + eprintln!("skipping: no non-loopback IPv4 address"); + return; + }; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + let listener = TcpListener::bind((ip, 0)).await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let data = serde_json::json!({"network_policies":{"upstream":{ + "name":"upstream", "endpoints":[{"host":ip.to_string(),"port":port}], + "binaries":[{"path":"/usr/bin/python3"}] + }}}); + let engine = Arc::new( + native_identity_engine( + include_str!("../data/sandbox-policy.rego"), + &data.to_string(), + ) + .unwrap(), + ); + let identity = Ok(ContractBinaryIdentity { + executable: ContractExecutableIdentity { + path: executable_path("/usr/bin/python3"), + digest: Some("00".repeat(32).parse().unwrap()), + }, + ancestors: vec![], + cmdline_paths: vec![], + }); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut byte = [0u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte).await.unwrap(); + request.push(byte[0]); + } + assert!(request.starts_with(b"GET /mcp HTTP/1.1")); + // Accept the allowed request, then close before its response. + }); + let target = format!("http://{ip}:{port}/mcp"); + let request = format!("GET {target} HTTP/1.1\r\nHost: {ip}:{port}\r\n\r\n"); + let (mut app, client) = tokio::io::duplex(8192); + let mut client: ProxyClient = tokio::io::BufReader::new(Box::new(client)); + let (activity, mut events) = mpsc::channel(4); + Box::pin(handle_forward_proxy( + "GET", + &target, + request.as_bytes(), + request.len(), + &mut client, + Some(&identity), + None, + engine, + Arc::new(BinaryIdentityCache::new()), + Arc::new(AtomicU32::new(std::process::id())), + None, + AgentProposals::default(), + Arc::new(None), + Arc::new(None), + None, + None, + None, + Some(&activity), + None, + )) + .await + .unwrap(); + drop(client); + let mut response = String::new(); + app.read_to_string(&mut response).await.unwrap(); + assert!( + response.starts_with("HTTP/1.1 502 Bad Gateway\r\n"), + "{response}" + ); + assert!(response.contains("x-openshell-request-id:")); + assert!(response.contains("Connection: close\r\n")); + assert!( + !events.try_recv().unwrap().denied, + "allowed decision must survive upstream EOF" + ); + server.await.unwrap(); + }) + .await + .expect("forward EOF must yield a response"); + } + #[tokio::test] async fn plaintext_mcp_forwarding_preserves_initialization_and_selected_revision() { if !cfg!(target_os = "linux") { From c54804740ead9cb16bbf6b661a1771bbcdec9b1d Mon Sep 17 00:00:00 2001 From: Shiju Date: Fri, 6 Nov 2026 23:36:18 +0530 Subject: [PATCH 4/4] fix(proxy): correlate upstream failures with request IDs Record the response request ID in the HTTP upstream-failure event so operators can match client errors to structured and shorthand logs. Cover supplied and absent IDs in the existing event schema test. Signed-off-by: Shiju --- crates/openshell-supervisor-network/src/l7/relay.rs | 5 +++++ .../src/l7/relay/upstream_close_tests.rs | 12 +++++++++++- 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 0906805a4e..9ffda28a30 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -826,6 +826,7 @@ where fn build_upstream_http_failure_event( ctx: &L7EvalContext, method: &str, + request_id: Option<&str>, response_code: Option, ) -> openshell_ocsf::OcsfEvent { let mut event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) @@ -838,6 +839,9 @@ fn build_upstream_http_failure_event( .dst_endpoint(Endpoint::from_domain(&ctx.host, ctx.port)) .message("HTTP upstream connection closed or failed") .status_detail("upstream_http_transport_failure"); + if let Some(request_id) = request_id { + event = event.unmapped("request_id", request_id); + } if let Some(code) = response_code { event = event.http_response(openshell_ocsf::HttpResponse { code }); } @@ -869,6 +873,7 @@ where egress_emit!(build_upstream_http_failure_event( ctx, method, + request_id, response_code )); delivery diff --git a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs index 8e9de1e871..0994ba13cd 100644 --- a/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs +++ b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs @@ -236,10 +236,16 @@ async fn client_body_eof_is_not_an_upstream_failure() { #[test] fn upstream_failure_event_reports_502_without_request_secrets() { - for response_started in [false, true] { + let request_id = "9b8a7f6e-5d4c-4b3a-9f8e-7d6c5b4a3f2e"; + for (response_started, request_id) in [ + (false, Some(request_id)), + (true, Some(request_id)), + (false, None), + ] { let event = build_upstream_http_failure_event( &context(), "POST", + request_id, (!response_started).then_some(502), ); let json = event.to_json().unwrap(); @@ -247,6 +253,10 @@ fn upstream_failure_event_reports_502_without_request_secrets() { assert_eq!(json["status_detail"], "upstream_http_transport_failure"); assert_eq!(json["http_request"]["http_method"], "POST"); assert!(json["http_request"].get("url").is_none()); + assert_eq!( + json["unmapped"].get("request_id"), + request_id.map(serde_json::Value::from).as_ref() + ); if response_started { assert!(json.get("http_response").is_none()); } else {