diff --git a/crates/openshell-cli/src/ssh.rs b/crates/openshell-cli/src/ssh.rs index 19913bff0f..a576d5a788 100644 --- a/crates/openshell-cli/src/ssh.rs +++ b/crates/openshell-cli/src/ssh.rs @@ -3,6 +3,8 @@ //! SSH connection and proxy utilities. +mod upload; + use crate::color::Colorize; use crate::tls::{ GrpcClient, TlsOptions, grpc_client, owner_replica, retry_unrouted, routed_to, with_owner_hint, @@ -19,6 +21,9 @@ use openshell_core::proto::{ CreateSshSessionRequest, GetSandboxRequest, SandboxPhase, SshRelayTarget, TcpForwardFrame, TcpForwardInit, tcp_forward_init, }; +use openshell_core::upload::{ + DestinationKind as UploadDestinationKind, ExpectedDestination, UploadPlan, +}; use std::fs; use std::future::Future; use std::io::{IsTerminal, Write}; @@ -64,7 +69,6 @@ const CONNECT_CHILD_TERMINATION_TIMEOUT: Duration = Duration::from_secs(5); const SSH_TRANSPORT_FAILURE_EXIT_CODE: i32 = 255; const SYNC_RETRY_ATTEMPTS: usize = 4; const SYNC_RETRY_DELAY: Duration = Duration::from_secs(2); -const UPLOAD_DESTINATION_CHANGED_EXIT_CODE: i32 = 73; static DEFAULT_SSH_LOG_LEVEL: OnceLock<&'static str> = OnceLock::new(); /// Set the SSH log level derived from CLI verbosity before starting commands. @@ -1186,74 +1190,13 @@ fn local_upload_path_is_file_like(path: &Path) -> bool { }) } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum UploadDestinationKind { - Directory, - File, - Missing, -} - -impl UploadDestinationKind { - fn as_str(self) -> &'static str { - match self { - Self::Directory => "directory", - Self::File => "file", - Self::Missing => "missing", - } - } -} - -// Check every existing parent before interpreting a failed existence test as -// absence. An inaccessible directory, a file used as a parent, or a dangling -// parent symlink must fail inspection instead of selecting filename semantics. -// Only fixed tokens go to stdout; diagnostics remain on stderr. -const UPLOAD_DESTINATION_KIND_SH: &str = r#"upload_destination_kind() { - upload_path=$1 - upload_parent=$2 - while :; do - if [ -d "$upload_parent" ]; then - if [ ! -x "$upload_parent" ]; then - printf 'cannot inspect upload destination parent: %s\n' "$upload_parent" >&2 - return 1 - fi - elif [ -e "$upload_parent" ] || [ -L "$upload_parent" ]; then - printf 'upload destination parent is not a directory: %s\n' "$upload_parent" >&2 - return 1 - fi - case "$upload_parent" in - /|.) break ;; - */*) upload_parent=${upload_parent%/*}; [ -n "$upload_parent" ] || upload_parent=/ ;; - *) upload_parent=. ;; - esac - done - if [ -d "$upload_path" ]; then - printf directory - elif [ -e "$upload_path" ] || [ -L "$upload_path" ]; then - printf file - else - printf missing - fi -}"#; - -fn upload_destination_kind_invocation(path: &str) -> String { - let (parent, _) = split_sandbox_path(path); - format!( - "upload_destination_kind {} {}", - shell_escape(path), - shell_escape(parent), - ) -} - fn file_upload_destination_needs_probe(path: &str) -> bool { // Keep the existing /name heuristic: a top-level path means a directory. !path.ends_with('/') && split_sandbox_path(path).0 != "/" } fn build_upload_destination_probe_cmd(path: &str) -> String { - format!( - "{UPLOAD_DESTINATION_KIND_SH}\n{}", - upload_destination_kind_invocation(path), - ) + openshell_core::upload::build_destination_probe(path) } fn parse_upload_destination_kind(output: &str) -> Result { @@ -1282,32 +1225,19 @@ fn resolve_file_upload_destination<'a>( } } +#[cfg(test)] fn build_upload_extract_cmd( dest_dir: &str, expected_destination: Option<(&str, UploadDestinationKind)>, ) -> String { - let escaped_dest = shell_escape(dest_dir); - let extract = format!("mkdir -p -- {escaped_dest} && cat | tar xf - -C {escaped_dest}"); - let Some((path, kind)) = expected_destination else { - return extract; - }; - - // Recheck before mkdir or tar can replace a path whose type changed after - // planning. This detects changes, but is not atomic with extraction and - // does not pin a symlink's target against concurrent replacement. - format!( - "{UPLOAD_DESTINATION_KIND_SH}\n\ - upload_kind=$({probe}) || exit 1\n\ - if [ \"$upload_kind\" != {expected} ]; then\n\ - printf 'upload destination changed type before extraction: %s\\n' {path} >&2\n\ - exit {changed_status}\n\ - fi\n\ - {extract}", - probe = upload_destination_kind_invocation(path), - expected = kind.as_str(), - path = shell_escape(path), - changed_status = UPLOAD_DESTINATION_CHANGED_EXIT_CODE, - ) + openshell_core::upload::build_extract_command(&UploadPlan { + destination_directory: dest_dir.to_string(), + expected_destination: expected_destination.map(|(path, kind)| ExpectedDestination { + path: path.to_string(), + kind, + }), + }) + .expect("test upload destination is valid") } /// Stream an archive into the sandbox and return its final destination. @@ -1335,8 +1265,16 @@ async fn ssh_tar_upload( { let output = ssh_run_capture_stdout(&session, &build_upload_destination_probe_cmd(path)) .await + .map_err(|error| { + if sync_error_is_retryable(&error) { + error + } else { + Report::new(upload::UploadError::Remote(error.to_string())) + } + }) .wrap_err_with(|| format!("failed to inspect upload destination '{path}'"))?; - let kind = parse_upload_destination_kind(&output)?; + let kind = parse_upload_destination_kind(&output) + .map_err(|error| Report::new(upload::UploadError::Protocol(error.to_string())))?; let (directory, basename) = resolve_file_upload_destination(path, tar_name, kind); dest_dir = directory; *tar_name = basename; @@ -1355,46 +1293,38 @@ async fn ssh_tar_upload( UploadSource::FileList { .. } => dest_dir.to_string(), }; - let mut ssh = ssh_base_command(&session.proxy_command); - ssh.arg("-T") + let ssh = ssh_upload_command(&session.proxy_command); + let plan = UploadPlan { + destination_directory: dest_dir.to_string(), + expected_destination: expected_destination.map(|(path, kind)| ExpectedDestination { + path: path.to_string(), + kind, + }), + }; + upload::run(ssh, plan, move |writer| { + write_upload_archive(writer, source) + }) + .await + .map_err(Report::new)?; + + Ok(final_path) +} + +/// Upload EOF marks a complete archive and the SSH exit status reports +/// extraction, so the session must stay in the process we spawn. User SSH +/// config may otherwise fork it into the background. +fn ssh_upload_command(proxy_command: &str) -> Command { + let mut command = ssh_base_command(proxy_command); + command + .arg("-o") + .arg("ForkAfterAuthentication=no") + .arg("-T") .arg("-o") .arg("RequestTTY=no") + .arg("-s") .arg("sandbox") - .arg(build_upload_extract_cmd(dest_dir, expected_destination)) - .stdin(Stdio::piped()) - .stdout(Stdio::null()) - .stderr(Stdio::inherit()); - - let mut child = ssh.spawn().into_diagnostic()?; - let stdin = child - .stdin - .take() - .ok_or_else(|| miette::miette!("failed to open stdin for ssh process"))?; - - // Wait even when the writer gets a broken pipe. A failed remote guard can - // close stdin early; its status must not become a retryable write error. - let archive_result = - tokio::task::spawn_blocking(move || -> Result<()> { write_upload_archive(stdin, source) }) - .await - .into_diagnostic()?; - let status = tokio::task::spawn_blocking(move || child.wait()) - .await - .into_diagnostic()? - .into_diagnostic()?; - - if status.code() == Some(UPLOAD_DESTINATION_CHANGED_EXIT_CODE) { - return Err(miette::miette!( - "upload destination changed type before extraction; retry the upload" - )); - } - if !status.success() { - return Err(miette::miette!( - "ssh tar extract exited with status {status}" - )); - } - archive_result?; - - Ok(final_path) + .arg(openshell_core::upload::SUBSYSTEM); + command } /// Split a sandbox path into (`parent_directory`, basename). @@ -1569,7 +1499,7 @@ pub async fn sandbox_sync_up_files( files: files.to_vec(), archive_prefix: file_list_archive_prefix(local_path), }; - retry_sandbox_sync("upload", || { + retry_sandbox_upload(|| { let source = source.clone(); async move { ssh_tar_upload(server, name, dest, source, tls, workspace).await } }) @@ -1609,7 +1539,7 @@ pub async fn sandbox_sync_up( local_path: local_path.to_path_buf(), tar_name, }; - retry_sandbox_sync("upload", || { + retry_sandbox_upload(|| { let source = source.clone(); async move { ssh_tar_upload(server, name, sandbox_path, source, tls, workspace).await } }) @@ -1656,9 +1586,10 @@ async fn ssh_run_capture_stdout(session: &SshSessionConfig, command: &str) -> Re .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::inherit()); - let output = tokio::task::spawn_blocking(move || ssh.output()) + let output = TokioCommand::from(ssh) + .kill_on_drop(true) + .output() .await - .into_diagnostic()? .into_diagnostic()?; if !output.status.success() { return Err(miette::miette!( @@ -1778,6 +1709,25 @@ async fn sandbox_sync_down_once( } } +async fn retry_sandbox_upload(run: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + #[cfg(unix)] + { + // Tokio retains handlers after listeners drop; cover setup and retry delays too. + let mut signals = TerminationSignals::new()?; + tokio::select! { + biased; + _ = signals.recv() => Err(Report::new(upload::UploadError::Cancelled)), + result = retry_sandbox_sync("upload", run) => result, + } + } + #[cfg(not(unix))] + retry_sandbox_sync("upload", run).await +} + async fn retry_sandbox_sync(operation: &str, mut run: F) -> Result where F: FnMut() -> Fut, @@ -1804,6 +1754,9 @@ where } fn sync_error_is_retryable(error: &Report) -> bool { + if let Some(error) = error.downcast_ref::() { + return error.is_retryable(); + } let message = format!("{error:?}").to_ascii_lowercase(); [ "broken pipe", @@ -2546,6 +2499,167 @@ mod tests { assert!(!sync_error_is_retryable(&error)); } + #[test] + fn upload_errors_retry_only_explicit_transport_failures() { + for error in [ + upload::UploadError::Unsupported, + upload::UploadError::Protocol("connection reset in invalid acknowledgment".into()), + upload::UploadError::Remote("destination /connection/unavailable".into()), + upload::UploadError::Archive("broken pipe while reading source".into()), + upload::UploadError::Cancelled, + ] { + assert!(!sync_error_is_retryable(&Report::new(error))); + } + assert!(sync_error_is_retryable(&Report::new( + upload::UploadError::Transport("SSH relay disconnected".into()) + ))); + } + + #[cfg(unix)] + #[test] + fn upload_cancellation_covers_retry_delay_and_setup() { + use std::os::unix::fs::PermissionsExt; + + const PHASE_ENV: &str = "OPENSHELL_UPLOAD_CANCEL_PHASE"; + const MARKER_ENV: &str = "OPENSHELL_UPLOAD_CANCEL_MARKER"; + const PID_ENV: &str = "OPENSHELL_UPLOAD_CANCEL_PROBE_PID"; + struct Probe(std::process::Child); + impl Drop for Probe { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + if let Some(phase) = std::env::var_os(PHASE_ENV) { + let marker = PathBuf::from(std::env::var_os(MARKER_ENV).unwrap()); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let mut attempts = 0; + let error = retry_sandbox_upload(|| { + attempts += 1; + assert_eq!(attempts, 1, "cancelled upload started another attempt"); + let marker = marker.clone(); + let setup = phase == "setup"; + async move { + if setup { + let session = SshSessionConfig { + proxy_command: String::new(), + sandbox_id: String::new(), + sandbox_name: String::new(), + gateway_url: String::new(), + token: String::new(), + main_terminal: false, + replica: None, + }; + return ssh_run_capture_stdout(&session, "unused").await.map(|_| ()); + } + let mut command = Command::new("sh"); + command.args(["-c", "exit 255"]); + let error = upload::run( + command, + UploadPlan { + destination_directory: "/tmp/upload".into(), + expected_destination: None, + }, + |_| Ok(()), + ) + .await + .unwrap_err(); + assert!(matches!(error, upload::UploadError::Transport(_))); + // A spawned task runs only after the current-thread + // runtime reaches the retry wait. + tokio::spawn(async move { + fs::write(marker, "retry delay").unwrap(); + }); + Err(Report::new(error)) + } + }) + .await + .unwrap_err(); + assert!(matches!( + error.downcast_ref::(), + Some(upload::UploadError::Cancelled) + )); + assert_eq!(attempts, 1); + if phase == "setup" { + let pid = fs::read_to_string(std::env::var_os(PID_ENV).unwrap()) + .unwrap() + .parse::() + .unwrap(); + tokio::time::timeout(Duration::from_millis(750), async { + while nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok() + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("cancelled probe must be terminated and reaped"); + } + }); + return; + } + + for phase in ["retry delay", "setup"] { + let temporary = tempfile::tempdir().unwrap(); + let marker = temporary.path().join("ready"); + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "ssh::tests::upload_cancellation_covers_retry_delay_and_setup", + "--nocapture", + ]) + .env(PHASE_ENV, phase) + .env(MARKER_ENV, &marker) + .env(PID_ENV, temporary.path().join("probe-pid")) + .stdin(Stdio::null()) + .stdout(Stdio::null()); + if phase == "setup" { + let ssh = temporary.path().join("ssh"); + fs::write( + &ssh, + "#!/bin/sh\n\ + printf '%s' \"$$\" > \"$OPENSHELL_UPLOAD_CANCEL_PROBE_PID\"\n\ + printf setup > \"$OPENSHELL_UPLOAD_CANCEL_MARKER\"\n\ + exec sleep 60\n", + ) + .unwrap(); + fs::set_permissions(ssh, fs::Permissions::from_mode(0o755)).unwrap(); + let mut path = temporary.path().as_os_str().to_os_string(); + path.push(":"); + path.push(std::env::var_os("PATH").unwrap_or_default()); + command.env("PATH", path); + } + let mut probe = Probe(command.spawn().unwrap()); + let deadline = Instant::now() + Duration::from_secs(5); + while !fs::read_to_string(&marker).is_ok_and(|ready| ready == phase) { + assert!(probe.0.try_wait().unwrap().is_none(), "probe exited early"); + assert!(Instant::now() < deadline, "probe did not reach {phase}"); + std::thread::sleep(Duration::from_millis(10)); + } + nix::sys::signal::kill( + nix::unistd::Pid::from_raw(probe.0.id().cast_signed()), + Signal::SIGINT, + ) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(1); + loop { + if let Some(status) = probe.0.try_wait().unwrap() { + assert!(status.success(), "upload cancellation failed: {status}"); + break; + } + assert!( + Instant::now() < deadline, + "cancellation waited through {phase}" + ); + std::thread::sleep(Duration::from_millis(10)); + } + } + } + #[test] fn connect_recovery_starts_only_for_established_transport_failures() { assert!(should_recover_connect( @@ -2804,6 +2918,21 @@ mod tests { ); } + #[cfg(unix)] + #[test] + fn upload_ssh_command_overrides_user_multiplexing_and_forking() { + let upload = ssh_upload_command("openshell ssh-proxy --sandbox demo"); + let effective = effective_ssh_config(&upload, &[]); + assert_connection_sharing_disabled(&effective); + for expected in [ + "forkafterauthentication no", + "requesttty false", + "sessiontype subsystem", + ] { + assert!(effective.lines().any(|line| line == expected), "{expected}"); + } + } + #[tokio::test] async fn wait_for_forward_listener_accepts_ready_listener() { let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) @@ -3456,7 +3585,7 @@ mod tests { let output = run_upload_shell(&command, tmp.path(), Some(&archive)); assert_eq!( output.status.code(), - Some(UPLOAD_DESTINATION_CHANGED_EXIT_CODE) + Some(openshell_core::upload::DESTINATION_CHANGED_EXIT_CODE) ); assert!(String::from_utf8_lossy(&output.stderr).contains("changed type")); if initial_kind == UploadDestinationKind::Directory { diff --git a/crates/openshell-cli/src/ssh/upload.rs b/crates/openshell-cli/src/ssh/upload.rs new file mode 100644 index 0000000000..05f234b9da --- /dev/null +++ b/crates/openshell-cli/src/ssh/upload.rs @@ -0,0 +1,693 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Bounded tar upload over the supervisor's credited SSH subsystem. + +use std::io::{self, Write}; +use std::process::{Command, ExitStatus, Stdio}; +use std::time::Duration; + +use miette::Diagnostic; +use openshell_core::upload::{ + ACK_INTERVAL_BYTES, DESTINATION_CHANGED_EXIT_CODE, UploadPlan, WINDOW_BYTES, +}; +use thiserror::Error; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::process::{Child, Command as TokioCommand}; +use tokio::sync::{mpsc, watch}; +use tokio::task::{AbortHandle, JoinHandle}; + +const NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(10); +const CHILD_EXIT_TIMEOUT: Duration = Duration::from_secs(5); +const CHUNK_BYTES: usize = 64 * 1024; +// A larger chunk could wait forever for credit the supervisor never reports. +const _: () = assert!(CHUNK_BYTES as u64 <= WINDOW_BYTES - ACK_INTERVAL_BYTES); +const MAX_DIAGNOSTIC_BYTES: usize = 16 * 1024; +/// The producer can hold at most two chunks in addition to its current buffer +/// and the chunk held by the sender. It cannot build an entire archive in RAM. +const PRODUCER_QUEUE_CHUNKS: usize = 2; + +#[derive(Debug, Error, Diagnostic)] +pub(super) enum UploadError { + #[error("upload SSH transport failed: {0}")] + Transport(String), + #[error( + "this sandbox supervisor does not support openshell-upload-v1; upgrade the sandbox supervisor before uploading" + )] + Unsupported, + #[error("upload protocol failed: {0}")] + Protocol(String), + #[error("sandbox upload failed: {0}")] + Remote(String), + #[error("failed to create upload archive: {0}")] + Archive(String), + #[error("upload cancelled")] + Cancelled, +} + +impl UploadError { + pub(super) fn is_retryable(&self) -> bool { + matches!(self, Self::Transport(_)) + } +} + +/// Own the SSH child through cancellation as well as ordinary error paths. +/// Tokio's kill-on-drop handles termination; the detached waiter also reaps it +/// when an enclosing command future is dropped by its caller. +struct ChildGuard(Option); + +impl ChildGuard { + fn child(&mut self) -> &mut Child { + self.0.as_mut().expect("SSH child is still owned") + } + + async fn finish(&mut self, terminate: bool) -> io::Result { + let child = self.child(); + if terminate { + child.start_kill()?; + } + let status = child.wait().await?; + self.0.take(); + Ok(status) + } +} + +impl Drop for ChildGuard { + fn drop(&mut self) { + if let Some(mut child) = self.0.take() { + let _ = child.start_kill(); + if let Ok(runtime) = tokio::runtime::Handle::try_current() { + runtime.spawn(async move { + let _ = child.wait().await; + }); + } + } + } +} + +struct AbortOnDrop(AbortHandle); + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } +} + +/// How a transfer ended before every archive byte was acknowledged. +#[derive(Debug)] +enum Stopped { + /// The SSH stream closed or failed; the exit status explains why. + Stream(String), + /// A local or protocol failure; the SSH child must be terminated. + Local(UploadError), +} + +/// Archive bytes sent and the final total the supervisor acknowledged. +#[derive(Debug)] +struct Sent { + bytes: u64, + acknowledged: u64, +} + +/// Start the producer only after the subsystem has started extraction. +/// The archive callback uses the existing tar builder unchanged. +pub(super) async fn run

( + mut command: Command, + plan: UploadPlan, + producer: P, +) -> Result<(), UploadError> +where + P: FnOnce(&mut dyn Write) -> miette::Result<()> + Send + 'static, +{ + let plan = plan + .encode() + .map_err(|error| UploadError::Protocol(error.to_string()))?; + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = TokioCommand::from(command) + .kill_on_drop(true) + .spawn() + .map_err(|error| UploadError::Transport(error.to_string()))?; + // EOF marks the end of the archive, so a failed or cancelled transfer must + // stop SSH before its stdin closes. The transfer closes stdin only on + // success, and declaring stdin first drops the guard, killing SSH, first. + let mut stdin = Some(child.stdin.take().expect("piped SSH stdin")); + let stdout = child.stdout.take().expect("piped SSH stdout"); + let stderr = child.stderr.take().expect("piped SSH stderr"); + let mut child = ChildGuard(Some(child)); + let diagnostics = tokio::spawn(read_diagnostic(stderr)); + let _diagnostics = AbortOnDrop(diagnostics.abort_handle()); + + let transfer = match transfer(stdout, &mut stdin, plan, producer).await { + Ok(sent) => Ok(sent), + Err(Stopped::Stream(message)) => Err(message), + Err(Stopped::Local(error)) => { + let _ = child.finish(true).await; + drop(stdin); + return Err(match error { + UploadError::Transport(message) => UploadError::Transport(format!( + "{message}: {}", + finish_diagnostic(diagnostics).await + )), + error => error, + }); + } + }; + drop(stdin); + let Ok(status) = tokio::time::timeout(CHILD_EXIT_TIMEOUT, child.finish(false)).await else { + let _ = child.finish(true).await; + return Err(UploadError::Transport( + "SSH did not exit after the upload stream ended".to_string(), + )); + }; + let status = status.map_err(|error| UploadError::Transport(error.to_string()))?; + let diagnostic = finish_diagnostic(diagnostics).await; + match status.code() { + Some(0) => match transfer { + Ok(sent) if sent.acknowledged == sent.bytes => Ok(()), + Ok(sent) => Err(UploadError::Protocol(format!( + "sandbox acknowledged {} of {} upload bytes", + sent.acknowledged, sent.bytes + ))), + Err(message) => Err(UploadError::Protocol(format!( + "upload stream ended before completion: {message}" + ))), + }, + // OpenSSH reports subsystem rejection as status 255, like other SSH + // failures. Its explicit request diagnostic distinguishes an old server + // from a transient connection failure, without sending any archive bytes. + Some(super::SSH_TRANSPORT_FAILURE_EXIT_CODE) + if diagnostic.contains("subsystem request failed") => + { + Err(UploadError::Unsupported) + } + Some(super::SSH_TRANSPORT_FAILURE_EXIT_CODE) | None => Err(UploadError::Transport( + format!("SSH exited with {status}: {diagnostic}"), + )), + Some(DESTINATION_CHANGED_EXIT_CODE) => Err(UploadError::Remote(format!( + "upload destination changed type before extraction; retry the upload: {diagnostic}" + ))), + Some(code) => Err(UploadError::Remote(format!( + "extraction exited with status {code}: {diagnostic}" + ))), + } +} + +/// The tail of SSH stderr, which includes extraction output. +async fn finish_diagnostic(task: JoinHandle) -> String { + tokio::time::timeout(CHILD_EXIT_TIMEOUT, task) + .await + .ok() + .and_then(Result::ok) + .unwrap_or_default() + .trim() + .to_string() +} + +async fn read_diagnostic(mut reader: R) -> String { + let mut diagnostic = Vec::new(); + let mut chunk = [0; 4096]; + loop { + match reader.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(bytes) => { + diagnostic.extend_from_slice(&chunk[..bytes]); + if diagnostic.len() > MAX_DIAGNOSTIC_BYTES { + diagnostic.drain(..diagnostic.len() - MAX_DIAGNOSTIC_BYTES); + } + } + } + } + String::from_utf8_lossy(&diagnostic).into_owned() +} + +enum ProducerEvent { + Data(Vec), + Finished(Result<(), UploadError>), +} + +struct ChunkWriter { + sender: mpsc::Sender, + buffer: Vec, +} + +impl ChunkWriter { + fn flush_chunk(&mut self) -> io::Result<()> { + if !self.buffer.is_empty() { + let buffer = std::mem::replace(&mut self.buffer, Vec::with_capacity(CHUNK_BYTES)); + self.sender + .blocking_send(ProducerEvent::Data(buffer)) + .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "upload stopped"))?; + } + Ok(()) + } +} + +impl Write for ChunkWriter { + fn write(&mut self, mut bytes: &[u8]) -> io::Result { + let original_len = bytes.len(); + while !bytes.is_empty() { + let length = bytes.len().min(CHUNK_BYTES - self.buffer.len()); + self.buffer.extend_from_slice(&bytes[..length]); + bytes = &bytes[length..]; + if self.buffer.len() == CHUNK_BYTES { + self.flush_chunk()?; + } + } + Ok(original_len) + } + + fn flush(&mut self) -> io::Result<()> { + self.flush_chunk() + } +} + +fn spawn_producer

(producer: P) -> mpsc::Receiver +where + P: FnOnce(&mut dyn Write) -> miette::Result<()> + Send + 'static, +{ + let (sender, receiver) = mpsc::channel(PRODUCER_QUEUE_CHUNKS); + tokio::task::spawn_blocking(move || { + let mut writer = ChunkWriter { + sender: sender.clone(), + buffer: Vec::with_capacity(CHUNK_BYTES), + }; + let result = producer(&mut writer) + .map_err(|error| UploadError::Archive(error.to_string())) + .and_then(|()| { + writer + .flush() + .map_err(|error| UploadError::Archive(error.to_string())) + }); + let _ = sender.blocking_send(ProducerEvent::Finished(result)); + }); + receiver +} + +/// Send the plan, wait for extraction to start, then stream the archive while +/// at most `WINDOW_BYTES` remain unacknowledged. Takes `stdin` only after the +/// archive is complete, so failures leave the caller to stop SSH first. +async fn transfer( + mut reader: R, + stdin: &mut Option, + plan: Vec, + producer: P, +) -> Result +where + R: AsyncRead + Unpin + Send + 'static, + W: AsyncWrite + Unpin + Send, + P: FnOnce(&mut dyn Write) -> miette::Result<()> + Send + 'static, +{ + let stream = |error: io::Error| Stopped::Stream(error.to_string()); + let writer = stdin.as_mut().expect("upload stdin is open"); + writer.write_all(&plan).await.map_err(stream)?; + writer.flush().await.map_err(stream)?; + match tokio::time::timeout(NEGOTIATION_TIMEOUT, reader.read_u64()).await { + Err(_) => { + return Err(Stopped::Local(UploadError::Transport( + "upload negotiation timed out".to_string(), + ))); + } + Ok(Err(error)) => return Err(stream(error)), + Ok(Ok(0)) => {} + Ok(Ok(_)) => { + return Err(Stopped::Local(UploadError::Protocol( + "upload did not start at offset zero".to_string(), + ))); + } + } + // The reader is never cancelled between partial records. + let (totals, mut acknowledged) = watch::channel(0_u64); + let acks = tokio::spawn(async move { + while let Ok(total) = reader.read_u64().await { + if total < *totals.borrow() { + return Err(UploadError::Protocol( + "upload acknowledgment moved backwards".to_string(), + )); + } + totals.send_replace(total); + } + Ok(()) + }); + let _acks = AbortOnDrop(acks.abort_handle()); + let mut chunks = spawn_producer(producer); + let mut sent = 0_u64; + loop { + let chunk = match chunks.recv().await { + Some(ProducerEvent::Data(chunk)) => chunk, + Some(ProducerEvent::Finished(result)) => { + result.map_err(Stopped::Local)?; + break; + } + None => { + return Err(Stopped::Local(UploadError::Archive( + "archive producer stopped".to_string(), + ))); + } + }; + let length = chunk.len() as u64; + loop { + let total = *acknowledged.borrow_and_update(); + if total > sent { + return Err(Stopped::Local(UploadError::Protocol( + "upload acknowledged unsent bytes".to_string(), + ))); + } + if sent + length - total <= WINDOW_BYTES { + break; + } + if acknowledged.changed().await.is_err() { + return Err(match acks.await { + Ok(Err(error)) => Stopped::Local(error), + _ => Stopped::Stream("upload acknowledgments ended".to_string()), + }); + } + } + sent += length; + writer.write_all(&chunk).await.map_err(stream)?; + } + // Closing stdin sends SSH EOF; the final total arrives before exit status. + writer.shutdown().await.map_err(stream)?; + stdin.take(); + if let Ok(Err(error)) = acks.await { + return Err(Stopped::Local(error)); + } + let acknowledged = *acknowledged.borrow(); + if acknowledged > sent { + return Err(Stopped::Local(UploadError::Protocol( + "upload acknowledged unsent bytes".to_string(), + ))); + } + Ok(Sent { + bytes: sent, + acknowledged, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use tokio::io::{AsyncBufReadExt, BufReader, DuplexStream, ReadHalf, WriteHalf}; + + #[allow(clippy::cast_possible_truncation)] + const WINDOW: usize = WINDOW_BYTES as usize; + type Check = fn(&Result<(), UploadError>) -> bool; + + fn plan() -> UploadPlan { + UploadPlan { + destination_directory: "/tmp/upload".to_string(), + expected_destination: None, + } + } + + fn start( + producer: impl FnOnce(&mut dyn Write) -> miette::Result<()> + Send + 'static, + buffer: usize, + ) -> (JoinHandle>, DuplexStream) { + let (client, server) = tokio::io::duplex(buffer); + let (read, write) = tokio::io::split(client); + let plan = plan().encode().unwrap(); + let transfer = tokio::spawn(async move { + let mut write = Some(write); + transfer(read, &mut write, plan, producer).await + }); + (transfer, server) + } + + /// Read the plan line and grant the zero acknowledgment. + async fn accept( + server: DuplexStream, + ) -> (BufReader>, WriteHalf) { + let (read, mut write) = tokio::io::split(server); + let mut read = BufReader::new(read); + let mut line = Vec::new(); + read.read_until(b'\n', &mut line).await.unwrap(); + assert_eq!(line, plan().encode().unwrap()); + write.write_u64(0).await.unwrap(); + (read, write) + } + + fn payload(bytes: usize) -> Vec { + (0..bytes) + .map(|index| u8::try_from(index % 251).unwrap()) + .collect() + } + + fn produce(bytes: Vec) -> impl FnOnce(&mut dyn Write) -> miette::Result<()> { + move |writer| writer.write_all(&bytes).map_err(miette::Report::from_err) + } + + #[tokio::test] + async fn stalled_consumer_blocks_at_credit_limit_then_resumes_with_exact_data() { + let expected = payload(6 * WINDOW + 123); + let (client, server) = start(produce(expected.clone()), 4096); + let (mut read, mut write) = accept(server).await; + let mut received = vec![0; WINDOW]; + read.read_exact(&mut received).await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(50), read.read_u8()) + .await + .is_err(), + "client exceeded credit while the extractor was stalled" + ); + assert!(!client.is_finished()); + // Acknowledge in ordinary-sized steps until the client sends EOF. + let mut chunk = vec![0; 256 * 1024]; + loop { + write.write_u64(received.len() as u64).await.unwrap(); + let size = read.read(&mut chunk).await.unwrap(); + if size == 0 { + break; + } + received.extend_from_slice(&chunk[..size]); + } + write.shutdown().await.unwrap(); + let sent = client.await.unwrap().unwrap(); + assert_eq!(received, expected); + assert_eq!(sent.bytes, expected.len() as u64); + assert_eq!(sent.acknowledged, sent.bytes); + } + + #[tokio::test] + async fn cancelling_stalled_transfer_stops_blocking_producer_and_closes_pipes() { + struct ProducerExit(Arc); + impl Drop for ProducerExit { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + let finished = Arc::new(AtomicBool::new(false)); + let producer_finished = finished.clone(); + let (client, server) = start( + move |writer| { + let _finished = ProducerExit(producer_finished); + writer + .write_all(&payload(8 * WINDOW)) + .map_err(miette::Report::from_err) + }, + 4096, + ); + let (mut read, _write) = accept(server).await; + let mut received = vec![0; WINDOW]; + read.read_exact(&mut received).await.unwrap(); + assert!(!finished.load(Ordering::SeqCst)); + client.abort(); + assert!(client.await.unwrap_err().is_cancelled()); + tokio::time::timeout(Duration::from_secs(2), async { + while !finished.load(Ordering::SeqCst) { + tokio::task::yield_now().await; + } + }) + .await + .expect("cancelled transfer must stop its blocking archive producer"); + let mut byte = [0]; + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), read.read(&mut byte)) + .await + .expect("cancelled transfer must release its protocol pipes") + .unwrap(), + 0 + ); + } + + #[tokio::test] + async fn impossible_and_regressing_acks_are_rejected_without_retry() { + for regress in [false, true] { + let (client, server) = start(produce(payload(4 * WINDOW)), 4096); + let (mut read, mut write) = accept(server).await; + if regress { + let mut chunk = vec![0; CHUNK_BYTES]; + read.read_exact(&mut chunk).await.unwrap(); + write.write_u64(CHUNK_BYTES as u64).await.unwrap(); + write.write_u64(CHUNK_BYTES as u64 - 1).await.unwrap(); + } else { + write.write_u64(u64::MAX).await.unwrap(); + } + // Keep consuming so the client reaches its next credit decision. + let drain = tokio::spawn(async move { + let _ = tokio::io::copy(&mut read, &mut tokio::io::sink()).await; + }); + let Err(Stopped::Local(error)) = client.await.unwrap() else { + panic!("invalid acknowledgment was accepted"); + }; + assert!(matches!(error, UploadError::Protocol(_))); + assert!(!error.is_retryable()); + drop(write); + drain.await.unwrap(); + } + } + + #[tokio::test] + async fn failed_archive_leaves_stdin_open_for_the_caller_to_stop_ssh() { + let (client, server) = tokio::io::duplex(4096); + let (read, write) = tokio::io::split(client); + let mut stdin = Some(write); + let server = tokio::spawn(async move { + let (mut read, write) = accept(server).await; + let mut partial = [0; 7]; + read.read_exact(&mut partial).await.unwrap(); + (read, write) + }); + let result = transfer(read, &mut stdin, plan().encode().unwrap(), |writer| { + writer + .write_all(b"partial") + .map_err(miette::Report::from_err)?; + writer.flush().map_err(miette::Report::from_err)?; + Err(miette::miette!("source file vanished")) + }) + .await; + assert!(matches!( + result, + Err(Stopped::Local(UploadError::Archive(_))) + )); + // EOF would mark a truncated archive as complete. + assert!(stdin.is_some()); + let (mut read, _write) = server.await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(50), read.read_u8()) + .await + .is_err() + ); + } + + #[tokio::test] + async fn closed_stream_interrupts_blocked_pipe_write() { + let (client, server) = start(produce(payload(2 * WINDOW)), 64); + let (read, write) = accept(server).await; + drop((read, write)); + let result = tokio::time::timeout(Duration::from_secs(1), client) + .await + .expect("remote closure must interrupt a blocked writer") + .unwrap(); + assert!(matches!(result, Err(Stopped::Stream(_)))); + } + + #[cfg(unix)] + async fn run_script(script: &str, started: Arc) -> Result<(), UploadError> { + let mut command = Command::new("sh"); + command.args(["-c", script]); + run(command, plan(), move |writer| { + started.store(true, Ordering::SeqCst); + writer + .write_all(b"archive") + .map_err(miette::Report::from_err) + }) + .await + } + + #[cfg(unix)] + #[tokio::test] + async fn exit_status_and_final_total_decide_the_result() { + // Each script reads the plan line, grants zero credit, then drains the + // archive before reporting. `\0\0\0\0\0\0\0\a` is a total of seven bytes. + const ZERO: &str = r"printf '\0\0\0\0\0\0\0\0'"; + const SEVEN: &str = r"printf '\0\0\0\0\0\0\0\a'"; + let cases: [(String, Check); 4] = [ + ( + format!("read -r plan; {ZERO}; cat >/dev/null; {SEVEN}"), + |result| result.is_ok(), + ), + (format!("read -r plan; {ZERO}; cat >/dev/null"), |result| { + matches!(result, Err(UploadError::Protocol(_))) + }), + ( + format!("read -r plan; {ZERO}; cat >/dev/null; echo 'tar: boom' >&2; exit 2"), + |result| matches!(result, Err(UploadError::Remote(message)) if message.contains("tar: boom")), + ), + ( + format!("read -r plan; {ZERO}; cat >/dev/null; exit 73"), + |result| matches!(result, Err(UploadError::Remote(message)) if message.contains("retry the upload")), + ), + ]; + for (script, check) in cases { + let started = Arc::new(AtomicBool::new(false)); + let result = run_script(&script, started.clone()).await; + assert!(check(&result), "{script}: {result:?}"); + assert!(started.load(Ordering::SeqCst)); + if let Err(error) = result { + assert!(!error.is_retryable()); + } + } + } + + #[cfg(unix)] + #[tokio::test] + async fn failure_before_extraction_starts_never_starts_archive_producer() { + for (script, retryable) in [ + ( + "printf 'subsystem request failed on channel 0\\n' >&2; exit 255", + false, + ), + ("exit 255", true), + ("read -r plan; echo 'mkdir: denied' >&2; exit 1", false), + ] { + let started = Arc::new(AtomicBool::new(false)); + let error = run_script(script, started.clone()).await.unwrap_err(); + assert!(!started.load(Ordering::SeqCst), "{script}"); + assert_eq!(error.is_retryable(), retryable, "{script}: {error:?}"); + if script.contains("subsystem") { + assert!(matches!(error, UploadError::Unsupported)); + } + } + } + + #[cfg(unix)] + #[tokio::test] + async fn dropping_upload_future_terminates_and_reaps_ssh_child() { + let temporary = tempfile::tempdir().unwrap(); + let pid_file = temporary.path().join("pid"); + let mut command = Command::new("sh"); + command.args([ + "-c", + "printf '%s' \"$$\" > \"$1\"; exec sleep 60", + "upload-test", + ]); + command.arg(&pid_file); + let upload = tokio::spawn(run(command, plan(), |_| Ok(()))); + let pid = tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Ok(pid) = tokio::fs::read_to_string(&pid_file).await + && let Ok(pid) = pid.parse::() + { + break pid; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + upload.abort(); + assert!(upload.await.unwrap_err().is_cancelled()); + tokio::time::timeout(Duration::from_secs(2), async { + while nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("SSH child must be terminated and reaped on cancellation"); + } +} diff --git a/crates/openshell-core/src/lib.rs b/crates/openshell-core/src/lib.rs index d07f9241a4..d4bfb4579a 100644 --- a/crates/openshell-core/src/lib.rs +++ b/crates/openshell-core/src/lib.rs @@ -62,6 +62,7 @@ pub mod spiffe; pub mod telemetry; pub mod time; pub mod transport_errors; +pub mod upload; pub use config::{ AppArmorProfile, Config, GatewayAuthConfig, GatewayInterceptorBindingOverride, diff --git a/crates/openshell-core/src/upload.rs b/crates/openshell-core/src/upload.rs new file mode 100644 index 0000000000..1483e1fe12 --- /dev/null +++ b/crates/openshell-core/src/upload.rs @@ -0,0 +1,251 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Internal streaming-upload protocol shared by the CLI and SSH supervisor. +//! +//! SSH receive credit is replenished before application consumption, so this +//! subsystem returns its own credit and relies on SSH for everything else. The +//! client sends one JSON plan line, the raw archive, then EOF. The supervisor's +//! stdout carries only big-endian `u64` totals of archive bytes written to +//! extraction stdin; the first total, zero, means extraction started. +//! Extraction output arrives on stderr and the SSH exit status is the result. + +use serde::{Deserialize, Serialize}; + +use crate::forward::shell_escape; + +pub const SUBSYSTEM: &str = "openshell-upload-v1"; +/// Archive bytes a client may send beyond the last acknowledged total. This +/// must stay below the supervisor's pending-stdin budget. +pub const WINDOW_BYTES: u64 = 1024 * 1024; +/// Archive bytes between progress reports; success also reports the final total. +/// +/// Clients must write archive chunks of at most +/// `WINDOW_BYTES - ACK_INTERVAL_BYTES`, so a drained extractor always returns +/// enough credit for the next chunk. +pub const ACK_INTERVAL_BYTES: u64 = 256 * 1024; +pub const MAX_PLAN_BYTES: usize = 16 * 1024; +pub const DESTINATION_CHANGED_EXIT_CODE: i32 = 73; + +#[derive(Debug, thiserror::Error)] +pub enum PlanError { + #[error("invalid upload plan: {0}")] + Invalid(&'static str), + #[error("invalid upload plan: {0}")] + Json(#[from] serde_json::Error), +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum DestinationKind { + Directory, + File, + Missing, +} + +impl DestinationKind { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Directory => "directory", + Self::File => "file", + Self::Missing => "missing", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExpectedDestination { + pub path: String, + pub kind: DestinationKind, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct UploadPlan { + pub destination_directory: String, + pub expected_destination: Option, +} + +impl UploadPlan { + pub fn validate(&self) -> Result<(), PlanError> { + validate_path(&self.destination_directory)?; + if let Some(expected) = &self.expected_destination { + validate_path(&expected.path)?; + } + Ok(()) + } + + /// Encode the newline-terminated plan that precedes the archive. JSON + /// escapes newlines inside strings, so the first newline ends the plan. + pub fn encode(&self) -> Result, PlanError> { + self.validate()?; + let mut line = serde_json::to_vec(self)?; + if line.len() > MAX_PLAN_BYTES { + return Err(PlanError::Invalid("plan exceeds limit")); + } + line.push(b'\n'); + Ok(line) + } +} + +fn validate_path(path: &str) -> Result<(), PlanError> { + if path.is_empty() || path.contains('\0') { + return Err(PlanError::Invalid("invalid destination path")); + } + Ok(()) +} + +/// Accumulates the plan line across SSH packets without exceeding the limit. +#[derive(Default)] +pub struct PlanReader(Vec); + +impl PlanReader { + /// Consume plan bytes from `input`, leaving any archive bytes after the + /// newline in place. Returns the plan once its line is complete. + pub fn read(&mut self, input: &mut &[u8]) -> Result, PlanError> { + let end = input.iter().position(|byte| *byte == b'\n'); + let line = &input[..end.unwrap_or(input.len())]; + if self.0.len() + line.len() > MAX_PLAN_BYTES { + return Err(PlanError::Invalid("plan exceeds limit")); + } + self.0.extend_from_slice(line); + let Some(end) = end else { + *input = &[]; + return Ok(None); + }; + *input = &input[end + 1..]; + let plan: UploadPlan = serde_json::from_slice(&self.0)?; + plan.validate()?; + Ok(Some(plan)) + } +} + +// This is the existing CLI destination probe, now shared with the receiving +// supervisor. Check existing parents before treating failed existence as absence. +const DESTINATION_KIND_SH: &str = r#"upload_destination_kind() { + upload_path=$1 + upload_parent=$2 + while :; do + if [ -d "$upload_parent" ]; then + if [ ! -x "$upload_parent" ]; then + printf 'cannot inspect upload destination parent: %s\n' "$upload_parent" >&2 + return 1 + fi + elif [ -e "$upload_parent" ] || [ -L "$upload_parent" ]; then + printf 'upload destination parent is not a directory: %s\n' "$upload_parent" >&2 + return 1 + fi + case "$upload_parent" in + /|.) break ;; + */*) upload_parent=${upload_parent%/*}; [ -n "$upload_parent" ] || upload_parent=/ ;; + *) upload_parent=. ;; + esac + done + if [ -d "$upload_path" ]; then + printf directory + elif [ -e "$upload_path" ] || [ -L "$upload_path" ]; then + printf file + else + printf missing + fi +}"#; + +fn destination_invocation(path: &str) -> String { + let parent = match path.rfind('/') { + Some(0) => "/", + Some(position) => &path[..position], + None => ".", + }; + format!( + "upload_destination_kind {} {}", + shell_escape(path), + shell_escape(parent) + ) +} + +#[must_use] +pub fn build_destination_probe(path: &str) -> String { + format!("{DESTINATION_KIND_SH}\n{}", destination_invocation(path)) +} + +pub fn build_extract_command(plan: &UploadPlan) -> Result { + plan.validate()?; + let dest = shell_escape(&plan.destination_directory); + let extract = format!("mkdir -p -- {dest} && cat | tar xf - -C {dest}"); + let Some(expected) = &plan.expected_destination else { + return Ok(extract); + }; + Ok(format!( + "{DESTINATION_KIND_SH}\n\ + upload_kind=$({probe}) || exit 1\n\ + if [ \"$upload_kind\" != {kind} ]; then\n\ + printf 'upload destination changed type before extraction: %s\\n' {path} >&2\n\ + exit {status}\n\ + fi\n\ + {extract}", + probe = destination_invocation(&expected.path), + kind = expected.kind.as_str(), + path = shell_escape(&expected.path), + status = DESTINATION_CHANGED_EXIT_CODE, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn plan() -> UploadPlan { + UploadPlan { + destination_directory: "/tmp/with 'quotes'\nand newline".into(), + expected_destination: Some(ExpectedDestination { + path: "/tmp/file".into(), + kind: DestinationKind::Missing, + }), + } + } + + #[test] + fn fragmented_plan_line_leaves_archive_bytes() { + let mut stream = plan().encode().unwrap(); + stream.extend_from_slice(b"archive\nbytes"); + for size in [1, 7, stream.len()] { + let mut reader = PlanReader::default(); + let mut decoded = None; + let mut rest = Vec::new(); + for chunk in stream.chunks(size) { + let mut input = chunk; + if decoded.is_none() { + decoded = reader.read(&mut input).unwrap(); + } + rest.extend_from_slice(input); + } + assert_eq!(decoded, Some(plan())); + assert_eq!(rest, b"archive\nbytes"); + } + } + + #[test] + fn plan_rejects_unknown_fields_nul_paths_and_unbounded_lines() { + let mut input = + &br#"{"destination_directory":"/tmp","expected_destination":null,"command":"id"} +"#[..]; + assert!(PlanReader::default().read(&mut input).is_err()); + let nul = UploadPlan { + destination_directory: "/tmp\0bad".into(), + expected_destination: None, + }; + assert!(nul.encode().is_err()); + let long = UploadPlan { + destination_directory: "/".repeat(MAX_PLAN_BYTES), + expected_destination: None, + }; + assert!(long.encode().is_err()); + let unterminated = vec![b'{'; MAX_PLAN_BYTES + 1]; + let mut reader = PlanReader::default(); + let (first, second) = unterminated.split_at(MAX_PLAN_BYTES); + assert!(reader.read(&mut &first[..]).unwrap().is_none()); + assert!(reader.read(&mut &second[..]).is_err()); + } +} diff --git a/crates/openshell-supervisor-process/README.md b/crates/openshell-supervisor-process/README.md new file mode 100644 index 0000000000..1c2f728b43 --- /dev/null +++ b/crates/openshell-supervisor-process/README.md @@ -0,0 +1,37 @@ +# OpenShell process supervisor + +This crate manages workload processes and SSH sessions through the admitted +isolation backend. + +## Bounded streaming uploads + +The CLI's built-in uploads use the `openshell-upload-v1` SSH subsystem. Russh +returns SSH receive credit before the supervisor consumes channel data, so this +subsystem returns its own credit and uses SSH for everything else. The gateway +relays the channel without interpreting it. + +The client sends one JSON plan line with the destination, then the raw tar +stream, then SSH EOF. When the plan arrives, the supervisor starts the +extractor through the ordinary `BoundaryExec` path, with the same workload +identity, environment, working directory and filesystem policy as other SSH +children. Startup runs inline, like an exec request. + +Stdout carries only big-endian `u64` totals of archive bytes that `InputTask` +has written through `BoundaryInput::write_all`. The first total, zero, means +extraction started. Extractor stdout and stderr are both sent as SSH stderr, +so extractor output cannot forge credit. The client keeps at most 1 MiB beyond +the last total outstanding; the supervisor rejects input beyond that window. +The existing 4 MiB pending-input guard remains active. + +After the extractor exits successfully and stdin has drained, the supervisor +sends the final total, then the exit status. The client succeeds only when the +status is zero and the final total equals the bytes it sent. A malformed plan, +credit violation, extended data or EOF before the plan fails the upload with +exit status 74. A rejection wins until the supervisor reports the exit result, +even after the extractor exits. Rejection, channel close or disconnect terminates the +extractor. Extraction writes directly to the destination, so failures can +leave partially extracted files; there is no rollback or atomic replacement. + +The CLI requires supervisor support for this subsystem before sending archive +data. It does not fall back to the previous unpaced exec-stdin upload path. +Ordinary exec stdin and SFTP retain their existing input-budget behavior. diff --git a/crates/openshell-supervisor-process/src/ssh.rs b/crates/openshell-supervisor-process/src/ssh.rs index 75c7b1fa5c..905a5f0a38 100644 --- a/crates/openshell-supervisor-process/src/ssh.rs +++ b/crates/openshell-supervisor-process/src/ssh.rs @@ -4,10 +4,14 @@ //! Embedded SSH server for sandbox access. mod input; +mod upload; #[cfg(test)] mod exec_input_tests; +#[cfg(test)] +mod upload_tests; + use crate::main_session::{MainOutput, MainSession}; #[cfg(unix)] use libc; @@ -371,6 +375,8 @@ async fn handle_connection( #[allow(clippy::struct_excessive_bools)] #[derive(Default)] struct ChannelState { + upload: Option, + command_started: bool, input_sender: Option, input_task: Option, process: Option>, @@ -607,6 +613,10 @@ impl russh::server::Handler for SshHandler { .channels .get_mut(&channel) .ok_or_else(|| anyhow::anyhow!("pty_request on unknown channel {channel:?}"))?; + if state.upload.is_some() || state.command_started { + session.channel_failure(channel)?; + return Ok(()); + } state.pty_request = Some(PtyRequest { term: term.to_string(), col_width, @@ -652,6 +662,14 @@ impl russh::server::Handler for SshHandler { channel: ChannelId, session: &mut Session, ) -> Result<(), Self::Error> { + if self + .channels + .get(&channel) + .is_some_and(|state| state.upload.is_some() || state.command_started) + { + session.channel_failure(channel)?; + return Ok(()); + } // Only allocate a PTY when the client explicitly requested one via // pty_request. VS Code Remote-SSH sends shell_request *without* a // preceding pty_request and expects pipe-based I/O with clean LF line @@ -674,6 +692,14 @@ impl russh::server::Handler for SshHandler { data: &[u8], session: &mut Session, ) -> Result<(), Self::Error> { + if self + .channels + .get(&channel) + .is_some_and(|state| state.upload.is_some() || state.command_started) + { + session.channel_failure(channel)?; + return Ok(()); + } let command = String::from_utf8_lossy(data).trim().to_string(); if command.is_empty() { session.channel_success(channel)?; @@ -698,6 +724,24 @@ impl russh::server::Handler for SshHandler { name: &str, session: &mut Session, ) -> Result<(), Self::Error> { + let Some(state) = self.channels.get_mut(&channel) else { + session.channel_failure(channel)?; + return Ok(()); + }; + if state.upload.is_some() || state.command_started || state.main_attached { + session.channel_failure(channel)?; + return Ok(()); + } + if name == openshell_core::upload::SUBSYSTEM { + if state.pty_request.is_some() { + session.channel_failure(channel)?; + return Ok(()); + } + // Extraction starts once the plan line arrives on stdin. + state.upload = Some(upload::Upload::new()); + session.channel_success(channel)?; + return Ok(()); + } if name == "openshell-main" { let Some(state) = self.channels.get(&channel) else { session.channel_failure(channel)?; @@ -881,6 +925,13 @@ impl russh::server::Handler for SshHandler { data: &[u8], session: &mut Session, ) -> Result<(), Self::Error> { + if self + .channels + .get(&channel) + .is_some_and(|state| state.upload.is_some()) + { + return self.upload_data(channel, data, session).await; + } let Some(state) = self.channels.get_mut(&channel) else { warn!("data on unknown channel {channel:?}"); return Ok(()); @@ -892,7 +943,7 @@ impl russh::server::Handler for SshHandler { if let Some(InputSender::Process(sender)) = &state.input_sender && sender.send(data) == Err(input::SendError::Full) { - self.reject_exec_input(channel, session)?; + self.reject_exec_input(channel, session, "process stdin buffer is full")?; } return Ok(()); } @@ -945,8 +996,16 @@ impl russh::server::Handler for SshHandler { async fn channel_eof( &mut self, channel: ChannelId, - _session: &mut Session, + session: &mut Session, ) -> Result<(), Self::Error> { + if self + .channels + .get(&channel) + .and_then(|state| state.upload.as_ref()) + .is_some_and(upload::Upload::awaiting_plan) + { + return self.reject_upload(channel, session, Some("upload ended before its plan")); + } // Drop only the sender: the writer drains accepted bytes and then // closes stdin. Commands such as `cat | tar xf -` need this half-close // to finish while their output remains available to the SSH client. @@ -969,6 +1028,23 @@ impl russh::server::Handler for SshHandler { Ok(()) } + async fn extended_data( + &mut self, + channel: ChannelId, + _ext: u32, + _data: &[u8], + session: &mut Session, + ) -> Result<(), Self::Error> { + if self + .channels + .get(&channel) + .is_some_and(|state| state.upload.is_some()) + { + return self.reject_upload(channel, session, None); + } + Ok(()) + } + async fn signal( &mut self, channel: ChannelId, @@ -1019,10 +1095,100 @@ impl russh::server::Handler for SshHandler { } impl SshHandler { + /// Read the plan, start extraction, then admit archive bytes within credit. + async fn upload_data( + &mut self, + channel: ChannelId, + mut data: &[u8], + session: &mut Session, + ) -> anyhow::Result<()> { + let Some(state) = self.channels.get_mut(&channel) else { + return Ok(()); + }; + let Some(upload) = state.upload.as_mut() else { + return Ok(()); + }; + if upload.awaiting_plan() { + let command = match upload.read_plan(&mut data).and_then(|plan| { + plan.map(|plan| openshell_core::upload::build_extract_command(&plan)) + .transpose() + }) { + Ok(None) => return Ok(()), + Ok(Some(command)) => command, + Err(error) => { + return self.reject_upload(channel, session, Some(&error.to_string())); + } + }; + let spec = openshell_isolation_interface::contract::ExecSpec { + program: String::new(), + args: Vec::new(), + shell: Some(openshell_isolation_interface::contract::ShellSpec { + command: Some(command), + login: !state.no_login_shell, + }), + runtime_helper: None, + env: Vec::new(), + workdir: None, + pty: false, + }; + // Like exec requests, start inline: a close is handled only after + // the process is registered with this channel. + if let Err(error) = self.start_exec_spec(channel, session.handle(), spec).await { + let message = format!("failed to start upload extraction: {error}"); + return self.reject_upload(channel, session, Some(&message)); + } + if data.is_empty() { + return Ok(()); + } + } + let Some(state) = self.channels.get_mut(&channel) else { + return Ok(()); + }; + let (Some(upload), Some(InputSender::Process(sender))) = + (state.upload.as_mut(), state.input_sender.as_ref()) + else { + return Ok(()); + }; + if !upload.admits(data.len()) { + return self.reject_upload(channel, session, None); + } + match sender.send(data) { + Ok(()) => upload.record(data.len()), + Err(input::SendError::Full) => return self.reject_upload(channel, session, None), + // The extractor closed stdin; its exit status reports the failure. + Err(input::SendError::Closed) => {} + } + Ok(()) + } + + /// Fail an upload with exit status 74 unless its result was reported. + fn reject_upload( + &mut self, + channel: ChannelId, + session: &mut Session, + message: Option<&str>, + ) -> anyhow::Result<()> { + if !self + .channels + .get(&channel) + .and_then(|state| state.upload.as_ref()) + .is_some_and(upload::Upload::reject) + { + return Ok(()); + } + // Only startup failures carry a message; the client's output window is + // still open then. Later rejections must not queue output. + if let Some(message) = message { + session.extended_data(channel, 1, format!("openshell: {message}\n").into_bytes())?; + } + self.reject_exec_input(channel, session, message.unwrap_or("invalid upload stream")) + } + fn reject_exec_input( &mut self, channel: ChannelId, session: &mut Session, + reason: &str, ) -> anyhow::Result<()> { if let Some(mut state) = self.channels.remove(&channel) { // A locally initiated close may never call channel_close. Release @@ -1031,7 +1197,7 @@ impl SshHandler { if let Some(process) = state.process.take() { tokio::spawn(async move { if let Err(error) = process.terminate().await { - warn!(%error, "failed to terminate exec after stdin overflow"); + warn!(%error, "failed to terminate rejected exec"); } }); } @@ -1039,7 +1205,7 @@ impl SshHandler { // Session methods enqueue directly. Do not add stderr data here: with // no output credit, russh would retain it and stop draining Handle // messages for other channels. Exit status needs no output credit. - warn!(?channel, "process stdin buffer is full; terminating exec"); + warn!(?channel, "{reason}; terminating exec"); session.exit_status_request(channel, 74)?; session.eof(channel)?; session.close(channel)?; @@ -1111,12 +1277,28 @@ impl SshHandler { .channels .get_mut(&channel) .ok_or_else(|| anyhow::anyhow!("exec on unknown channel {channel:?}"))?; + if state.upload.is_some() && exec.stdin.is_none() { + let process = exec.process.clone(); + tokio::spawn(async move { + let _ = process.terminate().await; + }); + anyhow::bail!("extraction process has no stdin"); + } state.process = Some(exec.process.clone()); + state.command_started = true; state.terminal = exec.terminal.take(); let output_status = exec.output_status.take(); + let mut finisher = None; if let Some(stdin) = exec.stdin.take() { - let (sender, task) = input::InputTask::spawn(stdin); + let (sender, task) = if let Some(upload) = state.upload.as_mut() { + let (sender, task, progress) = input::InputTask::spawn_with_progress(stdin); + finisher = + Some(upload.start(channel, handle.clone(), progress, exec.process.clone())); + (sender, task) + } else { + input::InputTask::spawn(stdin) + }; state.input_sender = Some(InputSender::Process(sender)); state.input_task = Some(task); } @@ -1125,6 +1307,8 @@ impl SshHandler { .as_ref() .map(input::InputTask::abort_handle); + // Upload stdout carries only credit, so extraction output uses stderr. + let stdout_extended = finisher.is_some(); let mut stdout = exec.stdout; let stdout_handle = handle.clone(); let stdout_task = tokio::spawn(async move { @@ -1133,7 +1317,12 @@ impl SshHandler { match stdout.read(&mut buffer).await { Ok(0) | Err(_) => break, Ok(size) => { - let _ = stdout_handle.data(channel, buffer[..size].to_vec()).await; + let data = buffer[..size].to_vec(); + let _ = if stdout_extended { + stdout_handle.extended_data(channel, 1, data).await + } else { + stdout_handle.data(channel, data).await + }; } } } @@ -1171,9 +1360,6 @@ impl SshHandler { } status }; - if let Some(input_abort) = input_abort { - input_abort.abort(); - } let code = match status { Some(openshell_isolation_interface::contract::BoundaryExitStatus::Exited(code)) => { code.max(0).cast_unsigned() @@ -1183,6 +1369,16 @@ impl SshHandler { )) => (128_i32.saturating_add(signal)).max(0).cast_unsigned(), None => 74, }; + let code = match finisher { + Some(finisher) => finisher.finish(&handle, code).await, + None => Some(code), + }; + if let Some(input_abort) = input_abort { + input_abort.abort(); + } + let Some(code) = code else { + return; + }; let _ = handle.eof(channel).await; let _ = handle.exit_status_request(channel, code).await; let _ = handle.close(channel).await; diff --git a/crates/openshell-supervisor-process/src/ssh/input.rs b/crates/openshell-supervisor-process/src/ssh/input.rs index 391fa62b70..46ca1dfa85 100644 --- a/crates/openshell-supervisor-process/src/ssh/input.rs +++ b/crates/openshell-supervisor-process/src/ssh/input.rs @@ -8,7 +8,7 @@ use std::sync::{Arc, Mutex}; use openshell_isolation_interface::contract::BoundaryInput; use tokio::io::AsyncWriteExt; -use tokio::sync::Notify; +use tokio::sync::{Notify, watch}; use tokio::task::{AbortHandle, JoinHandle}; /// Includes queued bytes and the chunk currently being written. Consumed bytes @@ -28,6 +28,22 @@ struct Buffer { struct Shared { buffer: Mutex, ready: Notify, + progress: Option>, +} + +/// A watch keeps only the newest consumed offset even when SSH output stalls. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct InputProgress { + pub(super) consumed: u64, + pub(super) state: InputState, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) enum InputState { + #[default] + Writing, + Drained, + Failed, } pub(super) struct InputSender(Arc); @@ -73,8 +89,30 @@ pub(super) struct InputTask { impl InputTask { pub(super) fn spawn(stdin: BoundaryInput) -> (InputSender, Self) { - let shared = Arc::new(Shared::default()); - let receiver = Receiver(shared.clone()); + Self::spawn_inner(stdin, None) + } + + pub(super) fn spawn_with_progress( + stdin: BoundaryInput, + ) -> (InputSender, Self, watch::Receiver) { + let (progress, receiver) = watch::channel(InputProgress::default()); + let (sender, task) = Self::spawn_inner(stdin, Some(progress)); + (sender, task, receiver) + } + + fn spawn_inner( + stdin: BoundaryInput, + progress: Option>, + ) -> (InputSender, Self) { + let shared = Arc::new(Shared { + progress, + ..Shared::default() + }); + let receiver = Receiver { + shared: shared.clone(), + consumed: 0, + finished: false, + }; let task = tokio::spawn(receiver.write_to(stdin)); (InputSender(shared.clone()), Self { shared, task }) } @@ -96,15 +134,28 @@ impl Drop for InputTask { } } -struct Receiver(Arc); +struct Receiver { + shared: Arc, + consumed: u64, + finished: bool, +} impl Receiver { - async fn write_to(self, mut stdin: BoundaryInput) { + fn publish(&self, state: InputState) { + if let Some(progress) = &self.shared.progress { + progress.send_replace(InputProgress { + consumed: self.consumed, + state, + }); + } + } + + async fn write_to(mut self, mut stdin: BoundaryInput) { loop { - let ready = self.0.ready.notified(); + let ready = self.shared.ready.notified(); let chunk = { - let Ok(mut buffer) = self.0.buffer.lock() else { - break; + let Ok(mut buffer) = self.shared.buffer.lock() else { + return; }; if buffer.bytes.is_empty() && buffer.closed { break; @@ -118,12 +169,22 @@ impl Receiver { continue; } if stdin.write_all(&chunk).await.is_err() { - break; + return; } - let Ok(mut buffer) = self.0.buffer.lock() else { - break; - }; - buffer.in_flight = 0; + { + let Ok(mut buffer) = self.shared.buffer.lock() else { + return; + }; + buffer.in_flight = 0; + } + self.consumed += chunk.len() as u64; + self.publish(InputState::Writing); + } + // Ordinary exec retains its existing drop-to-close behavior. Upload + // needs an explicit shutdown result before claiming the drain complete. + if self.shared.progress.is_none() || stdin.shutdown().await.is_ok() { + self.finished = true; + self.publish(InputState::Drained); } } } @@ -132,11 +193,14 @@ impl Drop for Receiver { fn drop(&mut self) { // A failed or cancelled writer must refuse later input and release the // queue even if the SSH channel has not yet received a close packet. - if let Ok(mut buffer) = self.0.buffer.lock() { + if let Ok(mut buffer) = self.shared.buffer.lock() { buffer.closed = true; buffer.in_flight = 0; buffer.bytes = VecDeque::new(); } + if !self.finished { + self.publish(InputState::Failed); + } } } @@ -182,4 +246,79 @@ mod tests { .unwrap(); assert_eq!(received, vec![42; 8192]); } + + #[tokio::test] + async fn progress_is_earned_after_complete_writes_and_reports_drain() { + let (stdin, mut reader) = tokio::io::duplex(1); + let (sender, _task, mut progress) = InputTask::spawn_with_progress(Box::new(stdin)); + sender.send(&vec![42; WRITE_CHUNK]).unwrap(); + reader.read_u8().await.unwrap(); + assert_eq!(progress.borrow().consumed, 0); + drop(sender); + let mut remaining = Vec::new(); + reader.read_to_end(&mut remaining).await.unwrap(); + assert_eq!(remaining, vec![42; WRITE_CHUNK - 1]); + progress + .wait_for(|progress| progress.state == InputState::Drained) + .await + .unwrap(); + assert_eq!(progress.borrow().consumed, WRITE_CHUNK as u64); + } + + #[tokio::test] + async fn failed_writer_publishes_terminal_state_and_refuses_input() { + let (stdin, reader) = tokio::io::duplex(1); + let (sender, _task, mut progress) = InputTask::spawn_with_progress(Box::new(stdin)); + drop(reader); + sender.send(b"write fails").unwrap(); + progress + .wait_for(|progress| progress.state == InputState::Failed) + .await + .unwrap(); + assert_eq!(progress.borrow().consumed, 0); + assert_eq!(sender.send(b"later input"), Err(SendError::Closed)); + } + + #[tokio::test] + async fn shutdown_failure_preserves_consumption_without_claiming_drain() { + struct ShutdownFails(tokio::io::DuplexStream); + + impl tokio::io::AsyncWrite for ShutdownFails { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + bytes: &[u8], + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.0).poll_write(cx, bytes) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.0).poll_flush(cx) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Err(std::io::Error::other("shutdown failed"))) + } + } + + let (stdin, mut reader) = tokio::io::duplex(1024); + let (sender, _task, mut progress) = + InputTask::spawn_with_progress(Box::new(ShutdownFails(stdin))); + sender.send(b"consumed before shutdown").unwrap(); + drop(sender); + let mut received = Vec::new(); + reader.read_to_end(&mut received).await.unwrap(); + assert_eq!(received, b"consumed before shutdown"); + progress + .wait_for(|progress| progress.state == InputState::Failed) + .await + .unwrap(); + assert_eq!(progress.borrow().consumed, received.len() as u64); + } } diff --git a/crates/openshell-supervisor-process/src/ssh/upload.rs b/crates/openshell-supervisor-process/src/ssh/upload.rs new file mode 100644 index 0000000000..c8656f8705 --- /dev/null +++ b/crates/openshell-supervisor-process/src/ssh/upload.rs @@ -0,0 +1,192 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Upload channels are exec channels whose stdin credit follows consumption. +//! +//! Russh replenishes SSH receive credit before `data()` runs, so the channel +//! reports how many archive bytes extraction stdin accepted. Stdout carries +//! only those totals; extraction output is sent as stderr so it cannot forge +//! credit. The SSH exit status reports the result. + +use super::input::{InputProgress, InputState}; +use openshell_core::upload::{ACK_INTERVAL_BYTES, PlanError, PlanReader, UploadPlan, WINDOW_BYTES}; +use openshell_isolation_interface::contract::BoundaryProcess; +use russh::ChannelId; +use russh::server::Handle; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tokio::sync::watch; +use tokio::task::{AbortHandle, JoinHandle}; + +// Honest clients cannot reach the general pending-input guard. +const _: () = assert!(WINDOW_BYTES < super::input::MAX_PENDING_INPUT as u64); +/// A successful `cat | tar` exit follows stdin EOF, so the writer publishes +/// its drained state almost immediately. +const DRAIN_TIMEOUT: Duration = Duration::from_secs(5); + +pub(super) struct Upload { + plan: Option, + received: u64, + progress: Option>, + process: Option>, + acks: Option, + /// Set by whichever side reports the result: the exit task or a rejection. + reported: Arc, +} + +impl Upload { + pub(super) fn new() -> Self { + Self { + plan: Some(PlanReader::default()), + received: 0, + progress: None, + process: None, + acks: None, + reported: Arc::new(AtomicBool::new(false)), + } + } + + pub(super) const fn awaiting_plan(&self) -> bool { + self.plan.is_some() + } + + /// Consume plan bytes, leaving archive bytes that follow it in `input`. + pub(super) fn read_plan(&mut self, input: &mut &[u8]) -> Result, PlanError> { + let plan = self + .plan + .as_mut() + .map_or(Ok(None), |reader| reader.read(input))?; + if plan.is_some() { + self.plan = None; + } + Ok(plan) + } + + /// Report extraction progress and return the exit task's finisher. + pub(super) fn start( + &mut self, + channel: ChannelId, + handle: Handle, + progress: watch::Receiver, + process: Arc, + ) -> Finisher { + let acks = tokio::spawn(send_acks( + handle, + channel, + progress.clone(), + process.clone(), + )); + self.acks = Some(acks.abort_handle()); + self.progress = Some(progress.clone()); + self.process = Some(process); + Finisher { + channel, + progress, + acks, + reported: self.reported.clone(), + } + } + + /// Whether `bytes` more fit in the credit granted to the client. + pub(super) fn admits(&self, bytes: usize) -> bool { + let consumed = self + .progress + .as_ref() + .map_or(0, |progress| progress.borrow().consumed); + self.received + bytes as u64 - consumed <= WINDOW_BYTES + } + + pub(super) fn record(&mut self, bytes: usize) { + self.received += bytes as u64; + } + + /// Claim the result for a rejection. False means the result was reported. + pub(super) fn reject(&self) -> bool { + !self.reported.swap(true, Ordering::AcqRel) + } +} + +impl Drop for Upload { + fn drop(&mut self) { + if let Some(acks) = self.acks.take() { + acks.abort(); + } + // Disconnects drop channel state without a close request. + if self.reject() + && let Some(process) = self.process.take() + { + tokio::spawn(async move { + let _ = process.terminate().await; + }); + } + } +} + +pub(super) struct Finisher { + channel: ChannelId, + progress: watch::Receiver, + acks: JoinHandle<()>, + reported: Arc, +} + +impl Finisher { + /// Send the final total for a successful extraction and return the exit + /// status to report, or `None` when a rejection already reported it. + pub(super) async fn finish(mut self, handle: &Handle, code: u32) -> Option { + self.acks.abort(); + let _ = (&mut self.acks).await; + let drained = code == 0 + && matches!( + tokio::time::timeout( + DRAIN_TIMEOUT, + self.progress.wait_for(|progress| progress.state != InputState::Writing), + ) + .await, + Ok(Ok(progress)) if progress.state == InputState::Drained + ); + if self.reported.swap(true, Ordering::AcqRel) { + return None; + } + if code != 0 { + return Some(code); + } + if !drained { + return Some(74); + } + let consumed = self.progress.borrow().consumed; + let _ = handle.data(self.channel, ack(consumed)).await; + Some(0) + } +} + +fn ack(consumed: u64) -> Vec { + consumed.to_be_bytes().to_vec() +} + +async fn send_acks( + handle: Handle, + channel: ChannelId, + mut progress: watch::Receiver, + process: Arc, +) { + // Zero tells the client that extraction started. + if handle.data(channel, ack(0)).await.is_err() { + return; + } + let mut acknowledged = 0; + while progress.changed().await.is_ok() { + let latest = *progress.borrow_and_update(); + if latest.state == InputState::Failed { + // The extractor stopped reading; its exit status explains why. + let _ = process.terminate().await; + return; + } + if latest.consumed - acknowledged >= ACK_INTERVAL_BYTES { + if handle.data(channel, ack(latest.consumed)).await.is_err() { + return; + } + acknowledged = latest.consumed; + } + } +} diff --git a/crates/openshell-supervisor-process/src/ssh/upload_tests.rs b/crates/openshell-supervisor-process/src/ssh/upload_tests.rs new file mode 100644 index 0000000000..196dccac79 --- /dev/null +++ b/crates/openshell-supervisor-process/src/ssh/upload_tests.rs @@ -0,0 +1,593 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Exercise the production upload handler over a real SSH session. + +use super::tests::{AcceptAnyServerKey, RejectingExec, test_client}; +use openshell_core::upload::{SUBSYSTEM, UploadPlan, WINDOW_BYTES}; +use openshell_isolation_interface::contract::{ + BackendError, BoundaryExec, BoundaryExitStatus, BoundaryProcess, BoundarySignal, ExecSession, + ExecSpec, +}; +use russh::ChannelMsg; +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; +use tokio::sync::{mpsc, oneshot, watch}; + +const DEADLINE: Duration = Duration::from_secs(10); +const CHUNK: usize = 64 * 1024; +#[allow(clippy::cast_possible_truncation)] +const WINDOW: usize = WINDOW_BYTES as usize; + +struct Process { + status: watch::Sender>, + waited: watch::Sender, + terminated: watch::Sender, + signals: mpsc::UnboundedSender, +} + +#[async_trait::async_trait] +impl BoundaryProcess for Process { + async fn wait(&self) -> Result { + let mut status = self.status.subscribe(); + let result = status + .wait_for(Option::is_some) + .await + .unwrap() + .as_ref() + .copied() + .unwrap(); + self.waited.send_replace(true); + Ok(result) + } + + async fn signal(&self, signal: BoundarySignal) -> Result<(), BackendError> { + let _ = self.signals.send(signal); + Ok(()) + } + + async fn terminate(&self) -> Result<(), BackendError> { + self.terminated.send_replace(true); + self.status + .send_replace(Some(BoundaryExitStatus::Exited(0))); + Ok(()) + } +} + +struct Exec(Mutex>); + +#[async_trait::async_trait] +impl BoundaryExec for Exec { + async fn exec(&self, spec: ExecSpec) -> Result { + assert!(!spec.pty); + assert!(spec.runtime_helper.is_none()); + if spec.shell.as_ref().unwrap().command.as_deref() != Some("cat") { + assert!(spec.shell.as_ref().unwrap().login); + } + Ok(self.0.lock().unwrap().pop_front().unwrap()) + } +} + +struct Control { + stdin: DuplexStream, + stdout: DuplexStream, + stderr: DuplexStream, + process: Arc, + signals: mpsc::UnboundedReceiver, +} + +impl Process { + async fn terminated(&self) { + self.terminated + .subscribe() + .wait_for(|terminated| *terminated) + .await + .unwrap(); + } +} + +fn exec_session() -> (ExecSession, Control) { + let (stdin, reader) = tokio::io::duplex(1024); + let (stdout, writer) = tokio::io::duplex(4096); + let (stderr, errors) = tokio::io::duplex(4096); + let (signal_tx, signals) = mpsc::unbounded_channel(); + let process = Arc::new(Process { + status: watch::channel(None).0, + waited: watch::channel(false).0, + terminated: watch::channel(false).0, + signals: signal_tx, + }); + ( + ExecSession { + process: process.clone(), + stdin: Some(Box::new(stdin)), + stdout: Box::new(stdout), + stderr: Some(Box::new(stderr)), + terminal: None, + output_status: None, + }, + Control { + stdin: reader, + stdout: writer, + stderr: errors, + process, + signals, + }, + ) +} + +async fn client_for(sessions: Vec) -> russh::client::Handle { + test_client( + None, + Arc::new(Exec(Mutex::new(sessions.into()))), + russh::client::Config::default(), + ) + .await +} + +type Channel = russh::Channel; + +/// Collects acknowledgment totals from stdout, stderr and exit statuses. +#[derive(Default)] +struct Reader { + stdout: Vec, + stderr: Vec, + statuses: Vec, +} + +impl Reader { + /// Next acknowledged total, or `None` once the channel closes. + async fn next(&mut self, channel: &mut Channel) -> Option { + loop { + if self.stdout.len() >= 8 { + let total = self.stdout.drain(..8).collect::>(); + return Some(u64::from_be_bytes(total.try_into().unwrap())); + } + match channel.wait().await? { + ChannelMsg::Data { data } => self.stdout.extend_from_slice(&data), + ChannelMsg::ExtendedData { data, ext: 1 } => self.stderr.extend_from_slice(&data), + ChannelMsg::ExitStatus { exit_status } => self.statuses.push(exit_status), + ChannelMsg::Close => return None, + ChannelMsg::WindowAdjusted { .. } | ChannelMsg::Success | ChannelMsg::Eof => {} + message => panic!("unexpected SSH message: {message:?}"), + } + } + } + + /// Remaining totals through channel close; no partial total may remain. + async fn close(&mut self, channel: &mut Channel) -> Vec { + let mut totals = Vec::new(); + while let Some(total) = self.next(channel).await { + totals.push(total); + } + assert!(self.stdout.is_empty(), "partial acknowledgment"); + totals + } +} + +fn plan_line() -> Vec { + UploadPlan { + destination_directory: "/tmp/upload".to_string(), + expected_destination: None, + } + .encode() + .unwrap() +} + +async fn open(client: &russh::client::Handle) -> Channel { + let mut channel = client.channel_open_session().await.unwrap(); + channel.request_subsystem(true, SUBSYSTEM).await.unwrap(); + assert!(matches!(channel.wait().await, Some(ChannelMsg::Success))); + channel +} + +async fn start(client: &russh::client::Handle) -> (Channel, Reader) { + let mut channel = open(client).await; + channel.data(plan_line().as_slice()).await.unwrap(); + let mut reader = Reader::default(); + assert_eq!(reader.next(&mut channel).await, Some(0)); + (channel, reader) +} + +#[tokio::test] +async fn stalled_upload_preserves_controls_then_resumes_beyond_four_mib() { + tokio::time::timeout(DEADLINE, async { + let (upload, mut control) = exec_session(); + let (sibling, mut other) = exec_session(); + let client = client_for(vec![upload, sibling]).await; + let (mut channel, mut reader) = start(&client).await; + let chunk = vec![42; CHUNK]; + let mut sent = 0; + while sent < WINDOW { + channel.data(chunk.as_slice()).await.unwrap(); + sent += CHUNK; + } + channel.signal(russh::Sig::INT).await.unwrap(); + assert_eq!(control.signals.recv().await, Some(BoundarySignal::Int)); + assert!( + tokio::time::timeout(Duration::from_millis(50), reader.next(&mut channel)) + .await + .is_err(), + "credit returned before the extractor consumed input" + ); + // Stalled upload input must leave sibling channel requests responsive. + let mut sibling = client.channel_open_session().await.unwrap(); + sibling.exec(true, "cat").await.unwrap(); + assert!(matches!(sibling.wait().await, Some(ChannelMsg::Success))); + sibling.data(&b"sibling"[..]).await.unwrap(); + sibling.eof().await.unwrap(); + let mut sibling_input = Vec::new(); + other.stdin.read_to_end(&mut sibling_input).await.unwrap(); + assert_eq!(sibling_input, b"sibling"); + sibling.close().await.unwrap(); + // Extractor stdout is delivered as stderr, so it cannot forge credit. + let forged = u64::MAX.to_be_bytes(); + control.stdout.write_all(&forged).await.unwrap(); + let noisy = tokio::spawn(async move { + control + .stderr + .write_all(&vec![9; 1024 * 1024]) + .await + .unwrap(); + }); + let consumed = tokio::spawn(async move { + let mut bytes = Vec::new(); + control.stdin.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, vec![42; 5 * WINDOW]); + drop(control.stdout); + control + .process + .status + .send_replace(Some(BoundaryExitStatus::Exited(0))); + }); + let mut acknowledged = 0; + while sent < 5 * WINDOW { + if sent + CHUNK > acknowledged + WINDOW { + let total = reader.next(&mut channel).await.unwrap(); + assert!(total >= acknowledged as u64 && total <= sent as u64); + acknowledged = usize::try_from(total).unwrap(); + } else { + channel.data(chunk.as_slice()).await.unwrap(); + sent += CHUNK; + } + } + channel.eof().await.unwrap(); + noisy.await.unwrap(); + let totals = reader.close(&mut channel).await; + assert_eq!(totals.last(), Some(&(sent as u64))); + assert_eq!(reader.statuses, [0]); + assert_eq!(reader.stderr.len(), forged.len() + 1024 * 1024); + assert!(reader.stderr.windows(8).any(|bytes| bytes == forged)); + consumed.await.unwrap(); + }) + .await + .expect("upload did not resume or blocked SSH control traffic"); +} + +#[tokio::test] +async fn small_upload_in_one_packet_reports_final_total() { + tokio::time::timeout(DEADLINE, async { + let (session, mut control) = exec_session(); + let client = client_for(vec![session]).await; + let mut channel = open(&client).await; + // The plan and archive share one packet; the tail stays below the + // progress interval, so only the final total covers it. + let mut packet = plan_line(); + packet.push(b'x'); + channel.data(packet.as_slice()).await.unwrap(); + channel.eof().await.unwrap(); + let mut bytes = Vec::new(); + control.stdin.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"x"); + control + .process + .status + .send_replace(Some(BoundaryExitStatus::Exited(0))); + drop((control.stdout, control.stderr)); + let mut reader = Reader::default(); + assert_eq!(reader.close(&mut channel).await, [0, 1]); + assert_eq!(reader.statuses, [0]); + }) + .await + .expect("small upload did not report its final total"); +} + +#[tokio::test] +async fn invalid_plans_and_credit_violations_exit_74() { + for case in ["credit", "eof", "plan"] { + tokio::time::timeout(DEADLINE, async { + let (session, control) = exec_session(); + let client = client_for(vec![session]).await; + let (mut channel, mut reader) = if case == "credit" { + start(&client).await + } else { + (open(&client).await, Reader::default()) + }; + match case { + "credit" => { + for _ in 0..=WINDOW / CHUNK { + channel.data(vec![7; CHUNK].as_slice()).await.unwrap(); + } + } + "eof" => { + channel.data(&b"{"[..]).await.unwrap(); + channel.eof().await.unwrap(); + } + _ => { + let plan = br#"{"destination_directory":"/tmp","expected_destination":null,"command":"id"} +"#; + channel.data(&plan[..]).await.unwrap(); + } + } + reader.close(&mut channel).await; + assert_eq!(reader.statuses, [74]); + if case == "credit" { + control.process.terminated().await; + } else { + let stderr = String::from_utf8_lossy(&reader.stderr); + assert!(stderr.contains("openshell: "), "{stderr}"); + assert!(!*control.process.terminated.borrow()); + } + }) + .await + .expect("rejected upload did not close"); + } +} + +#[tokio::test] +async fn extractor_failures_report_status_and_output() { + for early_exit in [true, false] { + tokio::time::timeout(DEADLINE, async { + let (session, mut control) = exec_session(); + let client = client_for(vec![session]).await; + let (mut channel, mut reader) = start(&client).await; + let diagnostic = "mkdir: /tmp/existing-file/child: Not a directory"; + if early_exit { + control + .stderr + .write_all(diagnostic.as_bytes()) + .await + .unwrap(); + control + .process + .status + .send_replace(Some(BoundaryExitStatus::Exited(2))); + } else { + // Broken stdin wakes no credit; the extractor is stopped. + drop(control.stdin); + channel.data(vec![7; CHUNK].as_slice()).await.unwrap(); + control.process.terminated().await; + } + drop((control.stdout, control.stderr)); + reader.close(&mut channel).await; + if early_exit { + assert_eq!(reader.statuses, [2]); + assert_eq!(reader.stderr, diagnostic.as_bytes()); + } else { + assert_eq!(reader.statuses, [74]); + } + }) + .await + .expect("extractor failure did not close the upload"); + } +} + +#[tokio::test] +async fn close_and_disconnect_cancel_stalled_boundary_input() { + for disconnect in [false, true] { + tokio::time::timeout(DEADLINE, async { + let (session, mut control) = exec_session(); + let client = client_for(vec![session]).await; + let (channel, _) = start(&client).await; + channel.data(vec![7; CHUNK].as_slice()).await.unwrap(); + channel.signal(russh::Sig::INT).await.unwrap(); + control.signals.recv().await.unwrap(); + if disconnect { + client + .disconnect(russh::Disconnect::ByApplication, "test", "") + .await + .unwrap(); + } else { + channel.close().await.unwrap(); + } + control.process.terminated().await; + let mut received = Vec::new(); + control.stdin.read_to_end(&mut received).await.unwrap(); + assert!(received.len() <= 1024); + }) + .await + .expect("close or disconnect stranded upload input"); + } +} + +#[tokio::test] +async fn upload_rejects_pty_and_duplicate_channel_requests() { + tokio::time::timeout(DEADLINE, async { + let client = test_client( + None, + Arc::new(RejectingExec), + russh::client::Config::default(), + ) + .await; + let mut pty = client.channel_open_session().await.unwrap(); + pty.request_pty(true, "xterm", 80, 24, 0, 0, &[]) + .await + .unwrap(); + assert!(matches!(pty.wait().await, Some(ChannelMsg::Success))); + pty.request_subsystem(true, SUBSYSTEM).await.unwrap(); + assert!(matches!(pty.wait().await, Some(ChannelMsg::Failure))); + let mut upload = open(&client).await; + upload.exec(true, "cat").await.unwrap(); + assert!(matches!(upload.wait().await, Some(ChannelMsg::Failure))); + upload.request_shell(true).await.unwrap(); + assert!(matches!(upload.wait().await, Some(ChannelMsg::Failure))); + upload.request_subsystem(true, "sftp").await.unwrap(); + assert!(matches!(upload.wait().await, Some(ChannelMsg::Failure))); + upload + .request_pty(true, "xterm", 80, 24, 0, 0, &[]) + .await + .unwrap(); + assert!(matches!(upload.wait().await, Some(ChannelMsg::Failure))); + upload.close().await.unwrap(); + }) + .await + .expect("duplicate upload request disrupted SSH session"); +} + +struct DelayedExec { + session: Mutex>, + started: watch::Sender, + release: Mutex>>, +} + +#[async_trait::async_trait] +impl BoundaryExec for DelayedExec { + async fn exec(&self, _: ExecSpec) -> Result { + let release = self.release.lock().unwrap().take().unwrap(); + self.started.send_replace(true); + release.await.unwrap(); + Ok(self.session.lock().unwrap().take().unwrap()) + } +} + +#[tokio::test] +async fn cancellation_during_remote_startup_terminates_late_process() { + for disconnect in [false, true] { + tokio::time::timeout(DEADLINE, async { + let (session, control) = exec_session(); + let (release, receiver) = oneshot::channel(); + let started = watch::channel(false).0; + let client = test_client( + None, + Arc::new(DelayedExec { + session: Mutex::new(Some(session)), + started: started.clone(), + release: Mutex::new(Some(receiver)), + }), + russh::client::Config::default(), + ) + .await; + let channel = open(&client).await; + channel.data(plan_line().as_slice()).await.unwrap(); + started + .subscribe() + .wait_for(|started| *started) + .await + .unwrap(); + // Startup holds the session loop, as for exec requests, so the + // close or disconnect is handled after the process is registered. + if disconnect { + client + .disconnect(russh::Disconnect::ByApplication, "test", "") + .await + .unwrap(); + } else { + channel.close().await.unwrap(); + } + release.send(()).unwrap(); + control.process.terminated().await; + }) + .await + .expect("cancelled startup stranded a late extractor"); + } +} + +#[tokio::test] +async fn late_rejection_wins_until_the_result_is_reported() { + for (after_exit, pending_output_status) in [(false, false), (true, false), (true, true)] { + tokio::time::timeout(DEADLINE, async { + let (mut session, mut control) = exec_session(); + let (output_status, receiver) = oneshot::channel(); + if pending_output_status { + session.output_status = Some(receiver); + } + let client = client_for(vec![session]).await; + let (mut channel, mut reader) = start(&client).await; + if after_exit { + channel.data(&b"x"[..]).await.unwrap(); + channel.eof().await.unwrap(); + let mut bytes = Vec::new(); + control.stdin.read_to_end(&mut bytes).await.unwrap(); + assert_eq!(bytes, b"x"); + if pending_output_status { + // Output EOF leaves the exit task waiting for the backend's + // final status; dropped readers prove both drains finished. + control.stdout.shutdown().await.unwrap(); + control.stderr.shutdown().await.unwrap(); + assert_eq!(control.stdout.read(&mut [0]).await.unwrap(), 0); + assert_eq!(control.stderr.read(&mut [0]).await.unwrap(), 0); + } else { + control + .process + .status + .send_replace(Some(BoundaryExitStatus::Exited(0))); + control + .process + .waited + .subscribe() + .wait_for(|waited| *waited) + .await + .unwrap(); + } + } + // The exit task is still waiting for output EOF or its final + // status, so the rejection must report without waiting for it. + channel + .extended_data(1, b"unexpected".as_slice()) + .await + .unwrap(); + assert!(reader.close(&mut channel).await.is_empty()); + assert_eq!(reader.statuses, [74]); + control.process.terminated().await; + drop((control.stdout, control.stderr, output_status)); + }) + .await + .expect("invalid upload input did not terminate"); + } +} + +struct LoginProbe(watch::Sender>); + +#[async_trait::async_trait] +impl BoundaryExec for LoginProbe { + async fn exec(&self, spec: ExecSpec) -> Result { + self.0.send_replace(Some(spec.shell.unwrap().login)); + Err(BackendError::Process("login selection observed".into())) + } +} + +#[tokio::test] +async fn upload_preserves_login_shell_selection() { + for no_login in [false, true] { + tokio::time::timeout(DEADLINE, async { + let login = watch::channel(None).0; + let client = test_client( + None, + Arc::new(LoginProbe(login.clone())), + russh::client::Config::default(), + ) + .await; + let mut channel = client.channel_open_session().await.unwrap(); + if no_login { + channel + .set_env(true, "OPENSHELL_NO_LOGIN_SHELL", "1") + .await + .unwrap(); + assert!(matches!(channel.wait().await, Some(ChannelMsg::Success))); + } + channel.request_subsystem(true, SUBSYSTEM).await.unwrap(); + assert!(matches!(channel.wait().await, Some(ChannelMsg::Success))); + channel.data(plan_line().as_slice()).await.unwrap(); + let mut reader = Reader::default(); + assert!(reader.close(&mut channel).await.is_empty()); + assert_eq!(reader.statuses, [74]); + let stderr = String::from_utf8_lossy(&reader.stderr); + assert!(stderr.contains("login selection observed"), "{stderr}"); + assert_eq!(*login.borrow(), Some(!no_login)); + }) + .await + .expect("login shell selection not applied to upload"); + } +} diff --git a/docs/how-it-works/sandboxes/overview.mdx b/docs/how-it-works/sandboxes/overview.mdx index 2e824c730e..2094a7280e 100644 --- a/docs/how-it-works/sandboxes/overview.mdx +++ b/docs/how-it-works/sandboxes/overview.mdx @@ -852,6 +852,16 @@ Upload files from your host into the sandbox: openshell sandbox upload my-sandbox ./src ``` +Uploads stream with bounded buffering and wait for the sandbox to consume +data, so the 4 MiB process-input budget does not limit the total upload size. +Upgrade both the CLI and the supervisor runtime used by your sandboxes to use +this behavior. The CLI reports an unsupported upload protocol before sending +file data to an older supervisor. + +Uploads extract directly into the destination. If you cancel an upload or it +fails, files already extracted remain and may be incomplete. Retry against the +existing sandbox after checking the destination, or remove the partial files. + When you omit the destination, OpenShell discovers the sandbox's working directory and uploads there. For a named local directory, OpenShell preserves the basename, matching `scp -r` and `cp -r`. If that directory already exists, diff --git a/e2e/rust/e2e-podman.sh b/e2e/rust/e2e-podman.sh index 0ebdc2359e..cd8b52e695 100755 --- a/e2e/rust/e2e-podman.sh +++ b/e2e/rust/e2e-podman.sh @@ -41,6 +41,7 @@ PODMAN_CI_TESTS=( podman_host_gateway podman_oci_identity provider_token_exchange + upload_backpressure ) if [ -n "${E2E_TEST}" ] && [ -n "${E2E_TEST_SET}" ]; then diff --git a/e2e/rust/tests/upload_backpressure.rs b/e2e/rust/tests/upload_backpressure.rs new file mode 100644 index 0000000000..e766303c37 --- /dev/null +++ b/e2e/rust/tests/upload_backpressure.rs @@ -0,0 +1,121 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#![cfg(all(feature = "e2e", unix))] + +//! Regression for large built-in uploads through the CLI, gateway, supervisor, +//! and sandbox. A fast sender previously overflowed the supervisor's 4 MiB +//! pending-stdin budget while tar was creating many files (issue #4430). + +use std::fmt::Write as _; +use std::fs; +use std::os::unix::fs::{PermissionsExt, symlink}; +use std::time::Duration; + +use openshell_e2e::harness::binary::openshell_cmd; +use openshell_e2e::harness::output::strip_ansi; +use openshell_e2e::harness::sandbox::SandboxGuard; +use rand::{RngCore, SeedableRng, rngs::StdRng}; +use sha2::{Digest, Sha256}; +use tokio::time::timeout; + +const FILE_COUNT: usize = 12_000; +const FILE_BYTES: usize = 8_192; +const TRANSFER_TIMEOUT: Duration = Duration::from_secs(300); + +/// An archive substantially larger than the stdin budget must finish without +/// pacing at the caller, preserving every file's content and tar semantics. +#[tokio::test] +async fn large_upload_preserves_content_and_metadata_outside_workdir() { + let tmpdir = tempfile::tempdir().expect("create upload fixture tempdir"); + let source = tmpdir.path().join("large-project"); + fs::create_dir_all(source.join("payload")).expect("create payload directory"); + + // Distinct deterministic pseudorandom bytes keep the fixture reproducible + // without letting compression or identical-file checks hide dropped data. + let mut rng = StdRng::seed_from_u64(4430); + let mut payload = [0_u8; FILE_BYTES]; + let mut checksums = String::new(); + for index in 0..FILE_COUNT { + rng.fill_bytes(&mut payload); + let relative = format!("payload/{index:05}.bin"); + fs::write(source.join(&relative), payload).expect("write payload file"); + writeln!(checksums, "{:x} {relative}", Sha256::digest(payload)) + .expect("append file checksum"); + } + let manifest_digest = format!("{:x}", Sha256::digest(checksums.as_bytes())); + fs::write(source.join("SHA256SUMS"), checksums).expect("write checksum manifest"); + fs::write( + source.join("run.sh"), + "#!/bin/sh\nprintf '%s\\n' upload-executable-ok\n", + ) + .expect("write executable fixture"); + fs::set_permissions(source.join("run.sh"), fs::Permissions::from_mode(0o751)) + .expect("set executable mode"); + symlink("payload/00000.bin", source.join("payload-link")).expect("create relative symlink"); + symlink("missing-target", source.join("dangling-link")).expect("create dangling symlink"); + fs::write(source.join("conflict.txt"), "local-conflict").expect("write merge fixture"); + + let mut sandbox = SandboxGuard::create(&[]) + .await + .expect("create persistent upload sandbox"); + sandbox + .exec(&[ + "sh", + "-c", + "set -eu; test \"$(pwd -P)\" != /tmp/openshell-upload-backpressure; \ + mkdir -p /tmp/openshell-upload-backpressure/large-project; \ + printf remote-conflict > /tmp/openshell-upload-backpressure/large-project/conflict.txt; \ + printf remote-preserved > /tmp/openshell-upload-backpressure/large-project/unrelated.txt", + ]) + .await + .expect("prepare destination outside the canonical workdir"); + + // No sleep, bandwidth limit, or retry: the protocol must backpressure the + // normal CLI tar stream, even when the local sender outruns extraction. + let mut upload = openshell_cmd(); + upload + .env("RUST_LOG", "warn") + .args(["sandbox", "upload", &sandbox.name]) + .arg(&source) + .args(["/tmp/openshell-upload-backpressure", "--no-git-ignore"]); + let result = timeout(TRANSFER_TIMEOUT, upload.output()) + .await + .expect("large upload must not hang waiting for credit") + .expect("start large upload"); + let output = format!( + "{}{}", + String::from_utf8_lossy(&result.stdout), + String::from_utf8_lossy(&result.stderr) + ); + assert!(result.status.success(), "large upload failed: {output}"); + assert!( + !output.contains("sandbox sync operation failed; retrying"), + "large upload required a retry: {output}" + ); + + let verify = format!( + "set -eu; cd /tmp/openshell-upload-backpressure/large-project; \ + test \"$(find payload -type f | wc -l)\" -eq {FILE_COUNT}; \ + test \"$(sha256sum SHA256SUMS | cut -d ' ' -f 1)\" = {manifest_digest}; \ + sha256sum --check --quiet SHA256SUMS; \ + test \"$(stat -c %a run.sh)\" = 751; \ + test \"$(./run.sh)\" = upload-executable-ok; \ + test -L payload-link; test \"$(readlink payload-link)\" = payload/00000.bin; \ + test -L dangling-link; test ! -e dangling-link; \ + test \"$(readlink dangling-link)\" = missing-target; \ + test \"$(cat conflict.txt)\" = local-conflict; \ + test \"$(cat unrelated.txt)\" = remote-preserved; \ + printf large-upload-content-and-metadata-ok" + ); + let output = timeout(TRANSFER_TIMEOUT, sandbox.exec(&["sh", "-c", &verify])) + .await + .expect("large upload verification must not hang") + .expect("every uploaded file and metadata item must match"); + assert!( + strip_ansi(&output).contains("large-upload-content-and-metadata-ok"), + "expected complete upload verification marker:\n{output}" + ); + + sandbox.cleanup().await; +} diff --git a/skills/openshell-cli/SKILL.md b/skills/openshell-cli/SKILL.md index 7bfb049f49..df2145dc5f 100644 --- a/skills/openshell-cli/SKILL.md +++ b/skills/openshell-cli/SKILL.md @@ -410,6 +410,8 @@ openshell sandbox upload my-sandbox ./src openshell sandbox download my-sandbox output ./local-output ``` +Uploads use bounded streaming and wait for sandbox input consumption; the 4 MiB process-input budget is not a total upload-size limit. Upgrade both the CLI and the supervisor runtime used by the sandbox. An older supervisor rejects the upload protocol before file data is sent. Failed or canceled uploads may leave incomplete files in the destination; check it before retrying or remove the partial files. + Uploads inside a Git work tree honor `.gitignore` by default and stop if Git filtering fails or selects no files. Sources confirmed to be outside a Git work tree upload without filtering, with a warning that `.gitignore` rules are not applied. Git must be available to determine whether filtering applies; broken or inaccessible repositories stop the upload. Add `--no-git-ignore` for an intentional unfiltered upload. This also applies to `sandbox create --upload`. `sandbox create --upload` checks every upload locally before provisioning and creates no sandbox if one is rejected. If a transfer fails after provisioning, the sandbox remains running and earlier uploads may have completed. Retry with `sandbox upload` against that sandbox, or remove it with `sandbox delete`.