diff --git a/src-tauri/src/services/remote_backend/bounded_output.rs b/src-tauri/src/services/remote_backend/bounded_output.rs new file mode 100644 index 000000000..1011c1cd6 --- /dev/null +++ b/src-tauri/src/services/remote_backend/bounded_output.rs @@ -0,0 +1,208 @@ +//! Allocation-bounded byte-to-line parsing for untrusted SSH output. +//! +//! `tokio::io::AsyncBufReadExt::lines` buffers until a newline before returning, +//! so a remote peer controls the size of that allocation. This module reads +//! fixed chunks and checks every byte budget before extending retained state. + +use tokio::io::{AsyncRead, AsyncReadExt}; + +const READ_CHUNK_BYTES: usize = 8 * 1024; + +#[derive(Clone, Copy, Debug)] +pub(crate) struct LineLimits { + pub max_line_bytes: usize, + pub max_stream_bytes: usize, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum LineLimitKind { + LineBytes, + StreamBytes, + ProtocolRecords, + RetainedBytes, + ProtocolEncoding, +} + +#[derive(Debug)] +pub(crate) enum BoundedLineError { + Io(std::io::Error), + Limit(LineLimitKind), +} + +impl std::fmt::Display for BoundedLineError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(error) => write!(f, "read failed: {error}"), + Self::Limit(LineLimitKind::LineBytes) => write!(f, "line byte limit exceeded"), + Self::Limit(LineLimitKind::StreamBytes) => write!(f, "stream byte limit exceeded"), + Self::Limit(LineLimitKind::ProtocolRecords) => { + write!(f, "protocol record limit exceeded") + } + Self::Limit(LineLimitKind::RetainedBytes) => { + write!(f, "retained byte limit exceeded") + } + Self::Limit(LineLimitKind::ProtocolEncoding) => { + write!(f, "protocol record is not valid UTF-8") + } + } + } +} + +impl std::error::Error for BoundedLineError {} + +/// Read `reader` as newline-delimited bytes and call `on_line` for each line. +/// +/// Lines exclude the LF delimiter and one optional preceding CR. A final +/// unterminated line is delivered at EOF. Input is not converted to UTF-8 until +/// after both the per-line and cumulative stream budgets have been enforced. +pub(crate) async fn read_bounded_lines( + mut reader: R, + limits: LineLimits, + mut on_line: F, +) -> Result<(), BoundedLineError> +where + R: AsyncRead + Unpin, + F: FnMut(&[u8]) -> Result<(), BoundedLineError>, +{ + read_bounded_lines_with_chunk_size(&mut reader, limits, READ_CHUNK_BYTES, &mut on_line).await +} + +async fn read_bounded_lines_with_chunk_size( + reader: &mut R, + limits: LineLimits, + chunk_bytes: usize, + on_line: &mut F, +) -> Result<(), BoundedLineError> +where + R: AsyncRead + Unpin, + F: FnMut(&[u8]) -> Result<(), BoundedLineError>, +{ + let mut chunk = vec![0_u8; chunk_bytes.clamp(1, READ_CHUNK_BYTES)]; + let mut line = Vec::with_capacity(limits.max_line_bytes.min(READ_CHUNK_BYTES)); + let mut stream_bytes = 0_usize; + + loop { + let read = reader + .read(&mut chunk) + .await + .map_err(BoundedLineError::Io)?; + if read == 0 { + if !line.is_empty() { + on_line(strip_cr(&line))?; + } + return Ok(()); + } + + stream_bytes = stream_bytes + .checked_add(read) + .filter(|total| *total <= limits.max_stream_bytes) + .ok_or(BoundedLineError::Limit(LineLimitKind::StreamBytes))?; + + let mut start = 0; + for (index, byte) in chunk[..read].iter().enumerate() { + if *byte != b'\n' { + continue; + } + extend_line(&mut line, &chunk[start..index], limits.max_line_bytes)?; + on_line(strip_cr(&line))?; + line.clear(); + start = index + 1; + } + extend_line(&mut line, &chunk[start..read], limits.max_line_bytes)?; + } +} + +fn extend_line( + line: &mut Vec, + bytes: &[u8], + max_line_bytes: usize, +) -> Result<(), BoundedLineError> { + line.len() + .checked_add(bytes.len()) + .filter(|total| *total <= max_line_bytes) + .ok_or(BoundedLineError::Limit(LineLimitKind::LineBytes))?; + line.extend_from_slice(bytes); + Ok(()) +} + +fn strip_cr(line: &[u8]) -> &[u8] { + line.strip_suffix(b"\r").unwrap_or(line) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn rejects_oversized_unterminated_line_before_growing_past_limit() { + let input = vec![b'x'; 65]; + let mut delivered = Vec::new(); + let error = read_bounded_lines( + input.as_slice(), + LineLimits { + max_line_bytes: 64, + max_stream_bytes: 1_024, + }, + |line| { + delivered.push(line.to_vec()); + Ok(()) + }, + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + BoundedLineError::Limit(LineLimitKind::LineBytes) + )); + assert!(delivered.is_empty()); + } + + #[tokio::test] + async fn rejects_many_small_lines_at_the_cumulative_stream_limit() { + let input = b"a\nb\nc\nd\n"; + let mut delivered = 0; + let error = read_bounded_lines( + input.as_slice(), + LineLimits { + max_line_bytes: 8, + max_stream_bytes: input.len() - 1, + }, + |_| { + delivered += 1; + Ok(()) + }, + ) + .await + .unwrap_err(); + + assert!(matches!( + error, + BoundedLineError::Limit(LineLimitKind::StreamBytes) + )); + assert_eq!(delivered, 0, "the single read is rejected before parsing"); + } + + #[tokio::test] + async fn preserves_utf8_split_across_fixed_reads() { + let input = b"a\xc3\xa9\n"; + let mut reader = input.as_slice(); + let mut lines = Vec::new(); + read_bounded_lines_with_chunk_size( + &mut reader, + LineLimits { + max_line_bytes: 8, + max_stream_bytes: 16, + }, + 2, + &mut |line| { + lines.push(String::from_utf8(line.to_vec()).unwrap()); + Ok(()) + }, + ) + .await + .unwrap(); + + assert_eq!(lines, ["aé"]); + } +} diff --git a/src-tauri/src/services/remote_backend/daemon.rs b/src-tauri/src/services/remote_backend/daemon.rs index 937feb429..703b86845 100644 --- a/src-tauri/src/services/remote_backend/daemon.rs +++ b/src-tauri/src/services/remote_backend/daemon.rs @@ -12,7 +12,9 @@ use std::time::Duration; use base64::Engine as _; use serde::Serialize; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncWriteExt, BufReader}; + +use super::bounded_output::{read_bounded_lines, BoundedLineError, LineLimitKind, LineLimits}; use super::error::{ classify_script_exit, classify_ssh_stderr, RemoteBackendError, RemoteBackendErrorKind, @@ -28,7 +30,18 @@ const BOOTSTRAP_SCRIPT: &str = include_str!("remote_daemon.sh"); // above their combined bounded lifetime so cancellation cannot strand an // unrecorded PID. const SCRIPT_TIMEOUT: Duration = Duration::from_secs(300); + +// `listdir` is the largest legal response: DIR + 2,000 entries + LIST-DONE. +// A 16 KiB encoded record covers a 4 KiB resolved path and filesystem +// components far beyond common 255-byte limits. Two MiB covers all 2,002 +// legal records with substantial encoding headroom. +const MAX_PROTOCOL_RECORDS: usize = 2_002; +const MAX_LINE_BYTES: usize = 16 * 1024; +const MAX_STDOUT_BYTES: usize = 4 * 1024 * 1024; +const MAX_PROTOCOL_BYTES: usize = 2 * 1024 * 1024; +const MAX_STDERR_STREAM_BYTES: usize = 1024 * 1024; const MAX_STDERR_BYTES: usize = 16 * 1024; +const MAX_DIAGNOSTIC_LOG_LINES: usize = 8; #[derive(Clone, Debug, Serialize)] #[serde(rename_all = "camelCase")] @@ -241,12 +254,142 @@ fn remote_command_line( line } +fn append_to_bounded_tail(tail: &mut String, text: &str, max_bytes: usize) { + if max_bytes == 0 { + return; + } + let line_budget = max_bytes - 1; + let mut start = text.len().saturating_sub(line_budget); + while !text.is_char_boundary(start) { + start += 1; + } + let text = &text[start..]; + let required = text.len() + 1; + if tail.len() + required > max_bytes { + let excess = tail.len() + required - max_bytes; + let mut drain_end = excess.min(tail.len()); + while !tail.is_char_boundary(drain_end) { + drain_end += 1; + } + tail.drain(..drain_end); + } + tail.push_str(text); + tail.push('\n'); +} + +fn stream_task_error(stream: &str, error: &BoundedLineError) -> RemoteBackendError { + let operation = if stream == "stdin" { "write" } else { "read" }; + let detail = match error { + BoundedLineError::Limit(_) => error.to_string(), + BoundedLineError::Io(source) => format!("{operation} failed: {source}"), + }; + let action = if stream == "stdin" { + "delivery failed" + } else { + "output rejected" + }; + RemoteBackendError::new( + RemoteBackendErrorKind::RemoteScriptFailed, + format!("ssh {stream} {action}: {detail}"), + ) +} + +struct ProtocolCollector { + lines: Vec, + retained_bytes: usize, +} + +impl ProtocolCollector { + fn new() -> Self { + Self { + lines: Vec::new(), + retained_bytes: 0, + } + } + + fn push(&mut self, protocol: &[u8]) -> Result<(), BoundedLineError> { + if self.lines.len() >= MAX_PROTOCOL_RECORDS { + return Err(BoundedLineError::Limit(LineLimitKind::ProtocolRecords)); + } + self.retained_bytes = self + .retained_bytes + .checked_add(protocol.len()) + .filter(|total| *total <= MAX_PROTOCOL_BYTES) + .ok_or(BoundedLineError::Limit(LineLimitKind::RetainedBytes))?; + let protocol = std::str::from_utf8(protocol) + .map_err(|_| BoundedLineError::Limit(LineLimitKind::ProtocolEncoding))?; + self.lines.push(protocol.to_owned()); + Ok(()) + } +} + +struct ScriptChildGuard(Option); + +enum WaitError { + Io(std::io::Error), + Timeout, +} + +impl ScriptChildGuard { + fn child_mut(&mut self) -> &mut tokio::process::Child { + self.0.as_mut().expect("script child already consumed") + } + + async fn wait_until( + mut self, + deadline: tokio::time::Instant, + ) -> Result { + let mut child = self.0.take().expect("script child already consumed"); + match child.try_wait() { + Ok(Some(status)) => return Ok(status), + Ok(None) => {} + Err(error) => return Err(WaitError::Io(error)), + } + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + match tokio::time::timeout(remaining, child.wait()).await { + Ok(Ok(status)) => Ok(status), + Ok(Err(error)) => Err(WaitError::Io(error)), + Err(_) => { + let _ = child.start_kill(); + let _ = child.wait().await; + Err(WaitError::Timeout) + } + } + } + + async fn terminate_and_reap(mut self) { + if let Some(mut child) = self.0.take() { + let _ = child.start_kill(); + let _ = child.wait().await; + } + } +} + +impl Drop for ScriptChildGuard { + fn drop(&mut self) { + if let Some(child) = self.0.as_mut() { + let _ = child.start_kill(); + } + } +} + pub(crate) async fn run_remote_script( spec: &RemoteHostSpec, shell_env: &HashMap, mode: &str, arg: Option<&str>, goose_arg: Option<&str>, +) -> Result { + run_remote_script_with_timeout(spec, shell_env, mode, arg, goose_arg, SCRIPT_TIMEOUT).await +} + +async fn run_remote_script_with_timeout( + spec: &RemoteHostSpec, + shell_env: &HashMap, + mode: &str, + arg: Option<&str>, + goose_arg: Option<&str>, + timeout: Duration, ) -> Result { let nonce = format!("berd-{}", uuid::Uuid::new_v4()); @@ -262,7 +405,7 @@ pub(crate) async fn run_remote_script( .stderr(Stdio::piped()) .kill_on_drop(true); - let mut child = command.spawn().map_err(|error| { + let child = command.spawn().map_err(|error| { if error.kind() == std::io::ErrorKind::NotFound { RemoteBackendError::new( RemoteBackendErrorKind::SshNotFound, @@ -272,77 +415,146 @@ pub(crate) async fn run_remote_script( RemoteBackendError::internal(format!("failed to spawn ssh: {error}")) } })?; + let mut child = ScriptChildGuard(Some(child)); let mut stdin = child + .child_mut() .stdin .take() .ok_or_else(|| RemoteBackendError::internal("ssh stdin unavailable"))?; let stdout = child + .child_mut() .stdout .take() .ok_or_else(|| RemoteBackendError::internal("ssh stdout unavailable"))?; let stderr = child + .child_mut() .stderr .take() .ok_or_else(|| RemoteBackendError::internal("ssh stderr unavailable"))?; - let run = async { + let deadline = tokio::time::Instant::now() + timeout; + let write_stdin = async { stdin .write_all(BOOTSTRAP_SCRIPT.as_bytes()) .await - .map_err(|error| { - RemoteBackendError::internal(format!("failed to send bootstrap script: {error}")) - })?; + .map_err(BoundedLineError::Io)?; drop(stdin); + Ok::<_, BoundedLineError>(()) + }; - let nonce_prefix = format!("{nonce} "); - let stdout_task = async { - let mut lines = Vec::new(); - let mut reader = BufReader::new(stdout).lines(); - while let Ok(Some(line)) = reader.next_line().await { - if let Some(protocol) = line.strip_prefix(&nonce_prefix) { - lines.push(protocol.to_string()); - } else if !line.trim().is_empty() { - log::debug!("[remote-backend noise] {}", redact_log_line(&line)); + let nonce_prefix = format!("{nonce} ").into_bytes(); + let stdout_task = async { + let mut collector = ProtocolCollector::new(); + let mut noise_lines = 0_usize; + read_bounded_lines( + BufReader::new(stdout), + LineLimits { + max_line_bytes: MAX_LINE_BYTES, + max_stream_bytes: MAX_STDOUT_BYTES, + }, + |line| { + if let Some(protocol) = line.strip_prefix(nonce_prefix.as_slice()) { + collector.push(protocol)?; + } else if !line.iter().all(u8::is_ascii_whitespace) { + if noise_lines < MAX_DIAGNOSTIC_LOG_LINES { + log::debug!( + "[remote-backend noise] {}", + redact_log_line(&String::from_utf8_lossy(line)) + ); + } + noise_lines = noise_lines.saturating_add(1); } - } - lines - }; - let stderr_task = async { - let mut collected = String::new(); - let mut reader = BufReader::new(stderr).lines(); - while let Ok(Some(line)) = reader.next_line().await { - let redacted = redact_log_line(&line); - log::warn!("[remote-backend ssh stderr] {redacted}"); - if collected.len() < MAX_STDERR_BYTES { - collected.push_str(&redacted); - collected.push('\n'); + Ok(()) + }, + ) + .await?; + if noise_lines > MAX_DIAGNOSTIC_LOG_LINES { + log::debug!( + "[remote-backend noise] suppressed {} additional lines", + noise_lines - MAX_DIAGNOSTIC_LOG_LINES + ); + } + Ok::<_, BoundedLineError>(collector.lines) + }; + let stderr_task = async { + let mut collected = String::new(); + let mut logged_lines = 0_usize; + read_bounded_lines( + BufReader::new(stderr), + LineLimits { + max_line_bytes: MAX_LINE_BYTES, + max_stream_bytes: MAX_STDERR_STREAM_BYTES, + }, + |line| { + let redacted = redact_log_line(&String::from_utf8_lossy(line)); + if logged_lines < MAX_DIAGNOSTIC_LOG_LINES { + log::warn!("[remote-backend ssh stderr] {redacted}"); } - } - collected - }; - - let (lines, stderr_text) = tokio::join!(stdout_task, stderr_task); - let status = child.wait().await.map_err(|error| { - RemoteBackendError::internal(format!("failed to await ssh: {error}")) - })?; - Ok::(ScriptOutput { - lines, - stderr: stderr_text, - exit_code: status.code(), - }) + logged_lines = logged_lines.saturating_add(1); + append_to_bounded_tail(&mut collected, &redacted, MAX_STDERR_BYTES); + Ok(()) + }, + ) + .await?; + if logged_lines > MAX_DIAGNOSTIC_LOG_LINES { + log::warn!( + "[remote-backend ssh stderr] suppressed {} additional lines", + logged_lines - MAX_DIAGNOSTIC_LOG_LINES + ); + } + Ok::<_, BoundedLineError>(collected) }; - match tokio::time::timeout(SCRIPT_TIMEOUT, run).await { - Ok(result) => result, - Err(_) => Err(RemoteBackendError::new( - RemoteBackendErrorKind::ReadyTimeout, - format!( - "ssh to {} timed out after {}s", - spec.destination(), - SCRIPT_TIMEOUT.as_secs() - ), - )), + let collected = tokio::time::timeout_at(deadline, async { + write_stdin.await.map_err(|error| ("stdin", error))?; + tokio::try_join!( + async { stdout_task.await.map_err(|error| ("stdout", error)) }, + async { stderr_task.await.map_err(|error| ("stderr", error)) }, + ) + }) + .await; + match collected { + Ok(Ok((lines, stderr_text))) => { + let status = match child.wait_until(deadline).await { + Ok(status) => status, + Err(WaitError::Io(error)) => { + return Err(RemoteBackendError::internal(format!( + "failed to await ssh: {error}" + ))) + } + Err(WaitError::Timeout) => { + return Err(RemoteBackendError::new( + RemoteBackendErrorKind::ReadyTimeout, + format!( + "ssh to {} timed out after {}s", + spec.destination(), + timeout.as_secs() + ), + )); + } + }; + Ok(ScriptOutput { + lines, + stderr: stderr_text, + exit_code: status.code(), + }) + } + Ok(Err((stream, error))) => { + child.terminate_and_reap().await; + Err(stream_task_error(stream, &error)) + } + Err(_) => { + child.terminate_and_reap().await; + Err(RemoteBackendError::new( + RemoteBackendErrorKind::ReadyTimeout, + format!( + "ssh to {} timed out after {}s", + spec.destination(), + timeout.as_secs() + ), + )) + } } } @@ -725,6 +937,129 @@ mod tests { assert!(require_success(&output(&["READY"], Some(0), "")).is_ok()); } + #[test] + fn stderr_tail_stays_bounded_and_retains_recent_diagnostics() { + let mut tail = String::new(); + append_to_bounded_tail(&mut tail, &"x".repeat(20), 16); + append_to_bounded_tail(&mut tail, "recent", 16); + assert!(tail.len() <= 16); + assert!(tail.ends_with("recent\n")); + } + + #[test] + fn protocol_record_limit_accepts_exact_boundary_and_rejects_one_more() { + let mut collector = ProtocolCollector::new(); + for _ in 0..MAX_PROTOCOL_RECORDS { + collector.push(b"E F eA==").unwrap(); + } + assert_eq!(collector.lines.len(), MAX_PROTOCOL_RECORDS); + assert!(matches!( + collector.push(b"LIST-DONE"), + Err(BoundedLineError::Limit(LineLimitKind::ProtocolRecords)) + )); + } + + #[test] + fn retained_protocol_limit_rejects_before_copy() { + let mut collector = ProtocolCollector::new(); + collector.retained_bytes = MAX_PROTOCOL_BYTES; + assert!(matches!( + collector.push(b"x"), + Err(BoundedLineError::Limit(LineLimitKind::RetainedBytes)) + )); + assert!(collector.lines.is_empty()); + } + + #[test] + fn invalid_protocol_encoding_is_not_reported_as_a_size_limit() { + let mut collector = ProtocolCollector::new(); + assert!(matches!( + collector.push(b"\xff"), + Err(BoundedLineError::Limit(LineLimitKind::ProtocolEncoding)) + )); + assert!(collector.lines.is_empty()); + } + + #[test] + fn stream_task_errors_identify_the_pipe_and_operation() { + let limit = BoundedLineError::Limit(LineLimitKind::StreamBytes); + assert_eq!( + stream_task_error("stdout", &limit).message, + "ssh stdout output rejected: stream byte limit exceeded" + ); + assert_eq!( + stream_task_error("stderr", &limit).message, + "ssh stderr output rejected: stream byte limit exceeded" + ); + let io = BoundedLineError::Io(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "closed", + )); + assert_eq!( + stream_task_error("stdin", &io).message, + "ssh stdin delivery failed: write failed: closed" + ); + } + + #[cfg(unix)] + #[tokio::test] + async fn elapsed_deadline_accepts_a_child_that_already_exited() { + let mut command = tokio::process::Command::new("sh"); + command.arg("-c").arg("exit 0").kill_on_drop(true); + let child = command.spawn().unwrap(); + let guard = ScriptChildGuard(Some(child)); + tokio::time::sleep(Duration::from_millis(50)).await; + + let result = guard + .wait_until(tokio::time::Instant::now() - Duration::from_millis(1)) + .await; + let status = match result { + Ok(status) => status, + Err(_) => panic!("an already-exited child must not be reported as timed out"), + }; + assert!(status.success()); + } + + #[tokio::test] + async fn timeout_kills_and_reaps_the_ssh_child() { + #[cfg(not(unix))] + return; + #[cfg(unix)] + { + let mut command = tokio::process::Command::new("sh"); + command.arg("-c").arg("sleep 120").kill_on_drop(true); + let child = command.spawn().unwrap(); + let pid = child.id().unwrap(); + let guard = ScriptChildGuard(Some(child)); + + let result = guard + .wait_until(tokio::time::Instant::now() + Duration::from_millis(50)) + .await; + assert!(matches!(result, Err(WaitError::Timeout))); + let status = std::process::Command::new("kill") + .args(["-0", &pid.to_string()]) + .status() + .unwrap(); + assert!(!status.success(), "timed-out ssh child {pid} survived"); + } + } + + #[test] + fn accepts_exact_maximum_listdir_protocol_envelope() { + let mut lines = Vec::with_capacity(MAX_PROTOCOL_RECORDS); + lines.push(format!("DIR {}", b64("/remote/path"))); + for index in 0..2_000 { + lines.push(format!("E F {}", b64(&format!("entry-{index}")))); + } + lines.push("LIST-DONE".to_string()); + + assert_eq!(lines.len(), MAX_PROTOCOL_RECORDS); + assert!(lines.iter().map(String::len).sum::() <= MAX_PROTOCOL_BYTES); + let listing = parse_dir_listing(&lines).unwrap(); + assert_eq!(listing.resolved_path, "/remote/path"); + assert_eq!(listing.entries.len(), 2_000); + } + #[test] fn secret_is_not_serialized() { let info = RemoteDaemonInfo { diff --git a/src-tauri/src/services/remote_backend/mod.rs b/src-tauri/src/services/remote_backend/mod.rs index f4b9c016e..d583130b4 100644 --- a/src-tauri/src/services/remote_backend/mod.rs +++ b/src-tauri/src/services/remote_backend/mod.rs @@ -20,6 +20,7 @@ //! Every state transition is emitted as [`REMOTE_BACKEND_STATUS_EVENT`] so the //! renderer can mirror per-host status without polling. +pub(crate) mod bounded_output; pub(crate) mod daemon; pub(crate) mod error; pub(crate) mod host; @@ -536,8 +537,7 @@ async fn establish( }) }; let Some(generation) = generation else { - let _ = tunnel.child.start_kill(); - let _ = tunnel.child.wait().await; + tunnel.terminate_and_reap().await; return Ok(None); }; emit_status(app, slot, generation, &state); @@ -569,14 +569,15 @@ async fn establish( fn spawn_supervisor( app: AppHandle, slot: Arc, - mut tunnel: tunnel::TunnelProcess, + tunnel: tunnel::TunnelProcess, generation: u64, prior_attempts: u32, ) { tauri::async_runtime::spawn(async move { let established_at = tokio::time::Instant::now(); + let mut tunnel = tunnel.into_parts(); let tunnel_pid = tunnel.child.id(); - let status = tunnel.child.wait().await; + let exit_detail = tunnel.wait_for_exit().await; let is_current = { let mut shared = slot.shared.lock().expect("slot poisoned"); @@ -588,10 +589,6 @@ fn spawn_supervisor( return; } - let exit_detail = match status { - Ok(status) => status.to_string(), - Err(error) => error.to_string(), - }; log::warn!( "[remote-backend] tunnel to {} closed unexpectedly ({exit_detail}); reconnecting", slot.key diff --git a/src-tauri/src/services/remote_backend/tunnel.rs b/src-tauri/src/services/remote_backend/tunnel.rs index 64ec7f163..a3d777024 100644 --- a/src-tauri/src/services/remote_backend/tunnel.rs +++ b/src-tauri/src/services/remote_backend/tunnel.rs @@ -11,8 +11,10 @@ use std::process::Stdio; use std::sync::{Arc, Mutex}; use std::time::Duration; -use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::{Child, Command}; +use tokio::task::JoinHandle; + +use super::bounded_output::{read_bounded_lines, BoundedLineError, LineLimits}; use super::error::{classify_ssh_stderr, RemoteBackendError, RemoteBackendErrorKind}; use super::host::RemoteHostSpec; @@ -21,12 +23,102 @@ use crate::services::log_redaction::redact_log_line; const TUNNEL_READY_TIMEOUT: Duration = Duration::from_secs(15); const TUNNEL_PROBE_INTERVAL: Duration = Duration::from_millis(150); +const TUNNEL_MAX_LINE_BYTES: usize = 16 * 1024; +const TUNNEL_MAX_STDERR_BYTES: usize = 1024 * 1024; +const TUNNEL_STDERR_TAIL_BYTES: usize = 8 * 1024; +const TUNNEL_MAX_LOG_LINES: usize = 8; + +type StderrTask = JoinHandle>; + +pub(crate) struct TunnelParts { + pub child: Child, + stderr_task: Option, +} + +impl TunnelParts { + /// Wait for an established tunnel to exit. The child owner—not the stderr + /// reader—terminates and reaps on reader failure, and the reader task is + /// always disposed before lifecycle/reconnect state may advance. + pub(crate) async fn wait_for_exit(&mut self) -> String { + let Some(mut stderr_task) = self.stderr_task.take() else { + return self + .child + .wait() + .await + .map(|status| status.to_string()) + .unwrap_or_else(|error| error.to_string()); + }; + + tokio::select! { + biased; + reader_result = &mut stderr_task => { + match reader_result { + Ok(Err(error)) => { + let _ = self.child.start_kill(); + let _ = self.child.wait().await; + format!("ssh tunnel output rejected: {error}") + } + Ok(Ok(())) => self + .child + .wait() + .await + .map(|status| status.to_string()) + .unwrap_or_else(|error| error.to_string()), + Err(error) => { + let _ = self.child.start_kill(); + let _ = self.child.wait().await; + format!("ssh tunnel stderr reader failed: {error}") + } + } + } + status = self.child.wait() => { + stderr_task.abort(); + let _ = stderr_task.await; + status.map(|status| status.to_string()).unwrap_or_else(|error| error.to_string()) + } + } + } +} pub(crate) struct TunnelProcess { pub child: Child, /// Rolling tail of redacted stderr, shared with the log-reader task, used /// to classify unexpected exits. pub stderr_tail: Arc>, + /// The reader reports output-limit violations to the child owner. It never + /// receives a PID or process-termination authority. + stderr_task: Option, +} + +impl TunnelProcess { + async fn take_finished_stderr_result( + &mut self, + ) -> Option, tokio::task::JoinError>> { + if !self + .stderr_task + .as_ref() + .is_some_and(JoinHandle::is_finished) + { + return None; + } + Some(self.stderr_task.take().expect("finished stderr task").await) + } + + pub(crate) async fn terminate_and_reap(&mut self) { + let _ = self.child.start_kill(); + let _ = self.child.wait().await; + if let Some(task) = self.stderr_task.take() { + task.abort(); + let _ = task.await; + } + } + + pub(crate) fn into_parts(mut self) -> TunnelParts { + TunnelParts { + child: self.child, + stderr_task: self.stderr_task.take(), + } + } } fn append_to_bounded_tail(tail: &mut String, line: &str, max_bytes: usize) { @@ -61,6 +153,41 @@ pub(crate) fn build_tunnel_command( command } +fn spawn_stderr_reader( + stderr: impl tokio::io::AsyncRead + Unpin + Send + 'static, + tail: Arc>, +) -> StderrTask { + tokio::spawn(async move { + let mut logged_lines = 0_usize; + let result = read_bounded_lines( + stderr, + LineLimits { + max_line_bytes: TUNNEL_MAX_LINE_BYTES, + max_stream_bytes: TUNNEL_MAX_STDERR_BYTES, + }, + |line| { + let redacted = redact_log_line(&String::from_utf8_lossy(line)); + if logged_lines < TUNNEL_MAX_LOG_LINES { + log::warn!("[remote-backend tunnel stderr] {redacted}"); + } + logged_lines = logged_lines.saturating_add(1); + if let Ok(mut tail) = tail.lock() { + append_to_bounded_tail(&mut tail, &redacted, TUNNEL_STDERR_TAIL_BYTES); + } + Ok::<_, BoundedLineError>(()) + }, + ) + .await; + if logged_lines > TUNNEL_MAX_LOG_LINES { + log::warn!( + "[remote-backend tunnel stderr] suppressed {} additional lines", + logged_lines - TUNNEL_MAX_LOG_LINES + ); + } + result + }) +} + pub(crate) fn spawn_tunnel( spec: &RemoteHostSpec, shell_env: &HashMap, @@ -86,21 +213,16 @@ pub(crate) fn spawn_tunnel( })?; let stderr_tail = Arc::new(Mutex::new(String::new())); - if let Some(stderr) = child.stderr.take() { - let tail = Arc::clone(&stderr_tail); - tauri::async_runtime::spawn(async move { - let mut lines = BufReader::new(stderr).lines(); - while let Ok(Some(line)) = lines.next_line().await { - let redacted = redact_log_line(&line); - log::warn!("[remote-backend tunnel stderr] {redacted}"); - if let Ok(mut tail) = tail.lock() { - append_to_bounded_tail(&mut tail, &redacted, 8 * 1024); - } - } - }); - } + let stderr_task = child + .stderr + .take() + .map(|stderr| spawn_stderr_reader(stderr, Arc::clone(&stderr_tail))); - Ok(TunnelProcess { child, stderr_tail }) + Ok(TunnelProcess { + child, + stderr_tail, + stderr_task, + }) } /// Probe the forwarded port until the remote goose answers HTTP. Any HTTP @@ -109,6 +231,26 @@ pub(crate) fn spawn_tunnel( pub(crate) async fn wait_for_tunnel_ready( local_port: u16, tunnel: &mut TunnelProcess, +) -> Result<(), RemoteBackendError> { + wait_for_tunnel_ready_with_timeout(local_port, tunnel, TUNNEL_READY_TIMEOUT).await +} + +async fn wait_for_tunnel_ready_with_timeout( + local_port: u16, + tunnel: &mut TunnelProcess, + ready_timeout: Duration, +) -> Result<(), RemoteBackendError> { + let result = wait_for_tunnel_ready_inner(local_port, tunnel, ready_timeout).await; + if result.is_err() { + tunnel.terminate_and_reap().await; + } + result +} + +async fn wait_for_tunnel_ready_inner( + local_port: u16, + tunnel: &mut TunnelProcess, + ready_timeout: Duration, ) -> Result<(), RemoteBackendError> { let client = reqwest::Client::builder() .timeout(Duration::from_secs(2)) @@ -116,8 +258,25 @@ pub(crate) async fn wait_for_tunnel_ready( .map_err(|error| RemoteBackendError::internal(format!("http client: {error}")))?; let url = format!("http://127.0.0.1:{local_port}/"); - let deadline = tokio::time::Instant::now() + TUNNEL_READY_TIMEOUT; + let deadline = tokio::time::Instant::now() + ready_timeout; loop { + if let Some(reader_result) = tunnel.take_finished_stderr_result().await { + match reader_result { + Ok(Err(error)) => { + return Err(RemoteBackendError::new( + RemoteBackendErrorKind::TunnelClosed, + format!("ssh tunnel output rejected: {error}"), + )); + } + Err(error) => { + return Err(RemoteBackendError::internal(format!( + "ssh tunnel stderr reader failed: {error}" + ))); + } + Ok(Ok(())) => {} + } + } + if let Some(status) = tunnel .child .try_wait() @@ -158,6 +317,7 @@ pub(crate) async fn wait_for_tunnel_ready( #[cfg(test)] mod tests { use super::*; + use crate::services::remote_backend::bounded_output::LineLimitKind; use crate::services::remote_backend::ssh::argv_of; #[test] @@ -186,4 +346,110 @@ mod tests { assert!(tail.len() <= 8 * 1024); assert_eq!(tail, format!("{}\n", "a".repeat(8_190))); } + + #[cfg(unix)] + fn local_tunnel_process(script: &str) -> TunnelProcess { + let mut command = Command::new("sh"); + command + .arg("-c") + .arg(script) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let mut child = command.spawn().unwrap(); + let stderr_tail = Arc::new(Mutex::new(String::new())); + let stderr_task = child + .stderr + .take() + .map(|stderr| spawn_stderr_reader(stderr, Arc::clone(&stderr_tail))); + TunnelProcess { + child, + stderr_tail, + stderr_task, + } + } + + #[cfg(unix)] + #[tokio::test] + async fn readiness_timeout_kills_reaps_and_joins_reader() { + let mut tunnel = local_tunnel_process("sleep 120"); + let result = + wait_for_tunnel_ready_with_timeout(9, &mut tunnel, Duration::from_millis(25)).await; + assert!(matches!( + result, + Err(RemoteBackendError { + kind: RemoteBackendErrorKind::ReadyTimeout, + .. + }) + )); + + assert!( + tunnel.child.id().is_none(), + "timed-out child was not reaped" + ); + assert!(tunnel.stderr_task.is_none(), "stderr reader was not joined"); + } + + #[cfg(unix)] + #[tokio::test] + async fn stderr_overflow_is_reported_to_owner_for_kill_and_reap() { + let mut tunnel = local_tunnel_process("yes overflow >&2"); + let result = wait_for_tunnel_ready(9, &mut tunnel).await; + let error = result.unwrap_err(); + assert_eq!(error.kind, RemoteBackendErrorKind::TunnelClosed); + assert!(error.message.contains("ssh tunnel output rejected")); + assert!( + tunnel.child.id().is_none(), + "overflowing child was not reaped" + ); + assert!(tunnel.stderr_task.is_none(), "stderr reader was not joined"); + } + + #[cfg(unix)] + #[tokio::test] + async fn completed_stderr_rejection_wins_over_simultaneous_child_exit() { + let mut command = tokio::process::Command::new("sh"); + command.arg("-c").arg("exit 0").kill_on_drop(true); + let child = command.spawn().unwrap(); + let stderr_task = + tokio::spawn(async { Err(BoundedLineError::Limit(LineLimitKind::StreamBytes)) }); + tokio::task::yield_now().await; + let mut tunnel = TunnelParts { + child, + stderr_task: Some(stderr_task), + }; + + let detail = tunnel.wait_for_exit().await; + + assert_eq!( + detail, + "ssh tunnel output rejected: stream byte limit exceeded" + ); + assert!(tunnel.child.id().is_none(), "exited child was not reaped"); + assert!(tunnel.stderr_task.is_none(), "stderr reader was not joined"); + } + + #[cfg(unix)] + #[tokio::test] + async fn established_stderr_overflow_is_owned_killed_reaped_and_joined() { + let tunnel = local_tunnel_process("yes overflow >&2"); + let mut tunnel = tunnel.into_parts(); + + let detail = tunnel.wait_for_exit().await; + + assert!(detail.contains("ssh tunnel output rejected")); + assert!( + detail.contains("stream byte limit exceeded"), + "unexpected detail: {detail}" + ); + assert!( + tunnel.child.id().is_none(), + "overflowing established child was not reaped" + ); + assert!( + tunnel.stderr_task.is_none(), + "established stderr reader was not joined" + ); + } }