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(