diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index fa20252714..9ffda28a30 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -21,6 +21,7 @@ use crate::l7::rest::WebSocketExtensionMode; use crate::l7::{EndpointObserver, EnforcementMode, L7EndpointConfig, L7Protocol, L7RequestInfo}; use crate::opa::{PolicyGenerationGuard, TunnelPolicyEngine}; use crate::telemetry::egress_emit; +use futures::FutureExt; use miette::{IntoDiagnostic, Result, miette}; use openshell_core::activity::{ActivitySender, try_record_activity}; use openshell_core::endpoint_status::{EndpointObservationSender, EndpointResult}; @@ -33,11 +34,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; @@ -758,6 +761,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,6 +823,62 @@ where .await } +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()) + .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 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 }); + } + 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); + egress_emit!(build_upstream_http_failure_event( + ctx, + method, + request_id, + response_code + )); + delivery +} + async fn relay_http_request_with_credential_rejection_observed( request: &crate::l7::provider::L7Request, client: &mut C, @@ -846,7 +906,19 @@ where { Ok(outcome) => Ok(Some(outcome)), Err(report) => { - if let Some(error) = report.downcast_ref::() { + if let Some(error) = report.downcast_ref::() { + handle_upstream_http_error( + client, + upstream, + ctx, + &request.action, + options.request_id, + error, + ) + .await?; + Ok(Some(RelayOutcome::Consumed)) + } 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. @@ -1216,25 +1288,26 @@ where .endpoint_observation_tx .as_ref() .and_then(EndpointObservationSender::capture); - let mut req = match provider.parse_request(client).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(()); @@ -2082,9 +2155,10 @@ 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.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!( @@ -2492,14 +2566,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 { @@ -2779,25 +2857,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 provider.parse_request(client).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?; @@ -3533,6 +3612,78 @@ 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( + request: F, + upstream: &mut U, + ctx: &L7EvalContext, +) -> Result> +where + F: Future>>, + U: AsyncRead + Unpin, +{ + let mut byte = [0u8; 1]; + let request = tokio::select! { + biased; + request = request => request, + result = upstream.read(&mut byte) => { + 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(); + egress_emit!(event); +} + /// Relay HTTP traffic with credential injection only (no L7 OPA evaluation). /// /// Used when TLS is auto-terminated but no L7 policy (`protocol` + `access`/`rules`) @@ -3567,9 +3718,10 @@ 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.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; @@ -3727,6 +3879,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 new file mode 100644 index 0000000000..0994ba13cd --- /dev/null +++ b/crates/openshell-supervisor-network/src/l7/relay/upstream_close_tests.rs @@ -0,0 +1,395 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::opa::{NetworkInput, OpaEngine}; +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"; + +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) +} + +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 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.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(); + 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_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 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(); + } + } + } + }) + .await + .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(); + relay_passthrough_with_credentials( + &mut client, + &mut relay_upstream, + &context(), + &guard, + Some(&engine), + ) + .await + .expect_err("client did not provide its declared body"); + 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() { + 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(); + 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()); + 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 { + assert_eq!(json["http_response"]["code"], 502); + } + 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 97c71f09b3..3442d08b0c 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,6 +832,33 @@ 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 { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("HTTP upstream connection closed or failed") + } +} + +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> { pub(crate) fn new( state: &'a openshell_core::provider_credentials::ProviderCredentialState, @@ -901,8 +930,60 @@ 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, + transport_failed: false, + read_eof: false, + }; + let mut client = ResponseProgress { + client, + response_started: false, + }; + let result = relay_http_exchange( + req, + &mut client, + &mut observed_upstream, + options, + response_middleware, + observer, + ) + .await; + // 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, + } +} + +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 @@ -1289,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 @@ -1296,6 +1423,18 @@ fn ensure_body_generation_current(options: RelayRequestOptions<'_>) -> Result<() struct ObservedUpstream<'a, U> { upstream: &'a mut U, observer: Option<&'a EndpointObserver>, + transport_failed: bool, + read_eof: 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 +1444,18 @@ impl AsyncRead for ObservedUpstream<'_, U> { buf: &mut tokio::io::ReadBuf<'_>, ) -> 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); - if matches!(result, std::task::Poll::Ready(Err(_))) - && let Some(observer) = this.observer - { - observer.observe(EndpointResult::TransportFailed); + 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))) + } + result => result, } - result } } @@ -1323,12 +1467,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 +1484,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( @@ -3200,9 +3347,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()?; @@ -3390,7 +3537,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 { @@ -3414,7 +3563,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(), }; @@ -6289,7 +6440,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 { @@ -6299,7 +6453,7 @@ mod tests { .expect("write response"); }); let mut client = tokio::io::sink(); - relay_response( + let result = relay_response( "POST", &mut upstream, &mut client, @@ -6309,14 +6463,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] @@ -6325,7 +6479,8 @@ mod tests { b"HTTP/1.1 204 No Content\r\n\r\n", EndpointResult::HttpResponseReceived, ) - .await; + .await + .unwrap(); } #[tokio::test] @@ -6334,7 +6489,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(); } } @@ -6413,7 +6570,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 + ); } } @@ -6482,6 +6647,8 @@ mod tests { let mut upstream = ObservedUpstream { upstream: &mut response, observer: Some(&observer), + transport_failed: false, + read_eof: false, }; let mut client = Vec::new(); relay_response( @@ -6514,7 +6681,8 @@ mod tests { b"not-http 200 nope\r\nContent-Length: 0\r\n\r\n", EndpointResult::TransportFailed, ) - .await; + .await + .unwrap(); } #[tokio::test] @@ -7339,6 +7507,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()); @@ -7365,14 +7534,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; @@ -7382,28 +7565,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 0113dc4323..26a4b0660f 100644 --- a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs +++ b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs @@ -95,10 +95,10 @@ 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 Ok(RelayOutcome::Consumed); + return Err(miette::Report::new(UpstreamHttpError { + response_started: informational_header_bytes > 0, + source: miette!("HTTP upstream closed before a complete response header"), + })); } buf.extend_from_slice(&tmp[..n]); }; @@ -1426,7 +1426,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); } @@ -1437,7 +1439,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 { @@ -1797,7 +1801,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 96c0b8f50a..0459b9a32d 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -5440,6 +5440,7 @@ where client, upstream, crate::l7::rest::RelayRequestOptions { + request_id: None, resolver: options.secret_resolver, body_classifier: options.body_classifier, mcp_request_validation: None, @@ -6916,27 +6917,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, }; @@ -9630,6 +9649,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") {