From 273c711cbde5b42d704efad0fdc949cd19a50dd0 Mon Sep 17 00:00:00 2001 From: Davanum Srinivas Date: Sat, 26 Sep 2026 22:37:41 -0400 Subject: [PATCH] fix(supervisor-network): keep workload bytes read with a mediated CONNECT header The proxy read the synthesized CONNECT header of a mediated open with a full-width read into an 8192-byte buffer through a BufReader of the same size, so the workload's request could arrive in the same read. The CONNECT path never looked past the header, and the request was lost until the workload timed out. Take the header out of the BufReader with fill_buf and consume only through the terminator, so the bytes behind it stay buffered for the relay. Signed-off-by: Davanum Srinivas --- .../openshell-supervisor-network/src/proxy.rs | 160 +++++++++++++++++- 1 file changed, 155 insertions(+), 5 deletions(-) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 5f224b474e..6ac6628cc4 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -2455,13 +2455,24 @@ async fn handle_mediated_connection( return Ok(()); } - let n = client.read(&mut buf[used..]).await.into_diagnostic()?; - if n == 0 { + // A mediated open's first workload bytes follow the synthesized CONNECT header. + // Take only the header out of the reader so the bytes behind it stay buffered for + // the relay; overlap three bytes so a terminator split across fills is found. + let available = client.fill_buf().await.into_diagnostic()?; + if available.is_empty() { return Ok(()); } - used += n; - - if buf[..used].windows(4).any(|win| win == b"\r\n\r\n") { + let n = available.len().min(buf.len() - used); + buf[used..used + n].copy_from_slice(&available[..n]); + let start = used.saturating_sub(3); + let end = buf[start..used + n] + .windows(4) + .position(|window| window == b"\r\n\r\n") + .map(|position| start + position + 4); + let consumed = end.map_or(n, |end| end - used); + client.consume(consumed); + used += consumed; + if end.is_some() { break; } } @@ -7660,6 +7671,145 @@ process: { run_as_user: sandbox, run_as_group: sandbox } assert!(response.contains("\"format\":\"yaml\""), "{response}"); } + #[tokio::test] + async fn mediated_connect_keeps_workload_bytes_read_with_the_synthesized_header() { + let Some(upstream_ip) = non_loopback_test_ipv4() else { + eprintln!("skipping: no routable non-loopback IPv4 test address"); + return; + }; + let upstream = TcpListener::bind((upstream_ip, 0)).await.unwrap(); + let destination = upstream.local_addr().unwrap(); + let (request_tx, request_rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + let (mut stream, _) = upstream.accept().await.unwrap(); + // Report whatever arrives before the request terminator or EOF, so a + // swallowed request shows up as an empty read, not as a hang. + let mut request = Vec::new(); + loop { + let mut byte = [0_u8; 1]; + match stream.read(&mut byte).await { + Ok(1) => request.push(byte[0]), + _ => break, + } + if request.ends_with(b"\r\n\r\n") { + break; + } + } + request_tx.send(request).unwrap(); + stream + .write_all(b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + }); + let engine = Arc::new( + OpaEngine::from_strings( + include_str!("../data/sandbox-policy.rego"), + &format!( + r#" +network_policies: + allowed: + name: allowed + endpoints: + - host: "{upstream_ip}" + port: {port} + tls: skip + binaries: + - path: /usr/bin/curl +"#, + port = destination.port() + ), + ) + .unwrap(), + ); + let cache = Arc::new(BinaryIdentityCache::new()); + let (stream, mut workload) = tokio::io::duplex(4096); + let (decision, completion) = tokio::sync::oneshot::channel(); + let pending = PendingTcpOpen { + stream: Box::new(stream), + binary_identity: Ok(ContractBinaryIdentity { + executable: ContractExecutableIdentity { + path: PathBuf::from("/usr/bin/curl"), + digest: Some("44".repeat(32).parse().unwrap()), + }, + ancestors: Vec::new(), + cmdline_paths: Vec::new(), + }), + destination, + socket: openshell_isolation_interface::contract::NetworkSocketMetadata { + socket_cookie: 7, + nonblocking: false, + process_generation: 1, + }, + policy_generation: engine.current_generation(), + timing: MediationTiming::default(), + decision, + }; + let (stream, supplied_identity, socket_addrs, transparent) = + preauthorize_transparent_open(pending, None, &engine, &cache, None, None, false, None) + .await + .expect("open is admitted"); + assert_eq!(completion.await.unwrap(), TcpOpenDecision::RelayReady); + // The workload writes before the proxy reads anything, so its request sits + // right behind the synthesized CONNECT header in the proxy's first read. + workload + .write_all( + format!("GET /ping HTTP/1.1\r\nHost: {destination}\r\nConnection: close\r\n\r\n") + .as_bytes(), + ) + .await + .unwrap(); + workload.shutdown().await.unwrap(); + let handler = tokio::spawn(handle_mediated_connection( + tokio::io::BufReader::new(stream), + supplied_identity, + socket_addrs, + transparent, + None, + engine, + cache, + Arc::new(AtomicU32::new(0)), + None, + None, + AgentProposals::new(true), + Arc::new(None), + Arc::new(None), + Arc::new(None), + None, + None, + None, + None, + None, + None, + )); + let request = tokio::time::timeout(std::time::Duration::from_secs(5), request_rx) + .await + .expect("upstream never received the workload's request") + .unwrap(); + assert!( + request.starts_with(b"GET /ping HTTP/1.1\r\n"), + "upstream received {:?} instead of the workload's request", + String::from_utf8_lossy(&request) + ); + let mut response = Vec::new(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + workload.read_to_end(&mut response), + ) + .await + .expect("response timed out") + .unwrap(); + assert!( + response.starts_with(b"HTTP/1.1 204 "), + "{}", + String::from_utf8_lossy(&response) + ); + tokio::time::timeout(std::time::Duration::from_secs(5), handler) + .await + .expect("handler did not finish") + .unwrap() + .unwrap(); + } + #[test] fn policy_local_route_requires_one_exact_host_and_an_origin_form_target() { assert!(valid_policy_local_request(