Files
OpenShell/crates/openshell-cli/src/ssh.rs
T
Daneyon Hansen 28add8924b fix(cli): preserve external TLS for SSH forwarding (#4344)
* fix(cli): preserve external TLS for SSH forwarding

Resolve the external scheme together with its authority when replacing an
internal bind address with the registered gateway endpoint.

Ported from #3714.

Fixes: #3674

Signed-off-by: Daneyon Hansen <daneyon.hansen@solo.io>

* docs(kubernetes): explain TLS SSH relay endpoint

Signed-off-by: Daneyon Hansen <daneyon.hansen@solo.io>

---------

Signed-off-by: Daneyon Hansen <daneyon.hansen@solo.io>
2026-10-12 20:13:15 +00:00

3758 lines
130 KiB
Rust

// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
//! SSH connection and proxy utilities.
use crate::color::Colorize;
use crate::tls::{
GrpcClient, TlsOptions, grpc_client, owner_replica, retry_unrouted, routed_to, with_owner_hint,
};
use miette::{IntoDiagnostic, Report, Result, WrapErr};
#[cfg(unix)]
use nix::sys::signal::Signal;
use openshell_core::driver_mounts;
use openshell_core::forward::{
ForwardSpec, build_proxy_command, format_gateway_url, resolve_ssh_gateway, sandbox_ssh_args,
shell_escape, validate_ssh_session_response, write_forward_pid,
};
use openshell_core::proto::{
CreateSshSessionRequest, GetSandboxRequest, SandboxPhase, SshRelayTarget, TcpForwardFrame,
TcpForwardInit, tcp_forward_init,
};
use std::fs;
use std::future::Future;
use std::io::{IsTerminal, Write};
#[cfg(unix)]
use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use std::process::{Command, ExitStatus, Stdio};
use std::sync::OnceLock;
use std::time::{Duration, Instant};
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use tokio::process::{Child, Command as TokioCommand};
use tokio_stream::wrappers::ReceiverStream;
use tonic::Code;
use tonic::metadata::AsciiMetadataValue;
/// Time budget for the local listener to become reachable after `ssh` starts.
/// This is a user-visible readiness deadline for both foreground and background
/// forwards, not a soft cleanup grace period.
const FORWARD_LISTENER_READINESS_TIMEOUT: Duration = Duration::from_secs(10);
/// Delay between listener/PID probes within the configured timeout.
const FORWARD_LISTENER_PROBE_INTERVAL: Duration = Duration::from_millis(50);
/// Per-attempt connect timeout, so one hung probe cannot consume the whole
/// grace period.
const FORWARD_LISTENER_CONNECT_TIMEOUT: Duration = Duration::from_millis(200);
/// Time budget for the supervisor relay to register after a fast canonical
/// command has already reported its terminal result.
const TERMINAL_RELAY_REGISTRATION_TIMEOUT: Duration = Duration::from_secs(5);
const TERMINAL_RELAY_REGISTRATION_INTERVAL: Duration = Duration::from_millis(50);
/// An SSH client that remained alive for this long was attached successfully,
/// rather than failing during initial authentication or setup.
const CONNECT_ESTABLISHED_DURATION: Duration = Duration::from_secs(2);
/// Briefly wait for the gateway lifecycle projection to catch up with an SSH
/// exit status before treating status 255 as a transport failure.
const CONNECT_TERMINAL_STATUS_TIMEOUT: Duration = Duration::from_millis(500);
const CONNECT_TERMINAL_STATUS_INTERVAL: Duration = Duration::from_millis(50);
/// Time allowed to restore an established canonical-main attachment after its
/// transport is lost.
const CONNECT_RECOVERY_TIMEOUT: Duration = Duration::from_mins(1);
const CONNECT_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(250);
const CONNECT_RETRY_MAX_DELAY: Duration = Duration::from_secs(5);
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.
pub fn set_default_ssh_log_level(level: &'static str) {
DEFAULT_SSH_LOG_LEVEL
.set(level)
.expect("SSH log level is configured only once");
}
#[derive(Clone, Copy, Debug)]
pub enum Editor {
Vscode,
Cursor,
}
impl Editor {
fn binary(self) -> &'static str {
match self {
Self::Vscode => "code",
Self::Cursor => "cursor",
}
}
fn remote_target(host_alias: &str) -> String {
format!("ssh-remote+{host_alias}")
}
fn label(self) -> &'static str {
match self {
Self::Vscode => "VS Code",
Self::Cursor => "Cursor",
}
}
}
struct SshSessionConfig {
proxy_command: String,
sandbox_id: String,
sandbox_name: String,
gateway_url: String,
token: String,
main_terminal: bool,
replica: Option<AsciiMetadataValue>,
}
async fn ssh_session_config(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
terminal_relay_registration_timeout: Option<Duration>,
) -> Result<SshSessionConfig> {
let mut client = grpc_client(server, tls).await?;
// Resolve the sandbox and retain its ID for local lifecycle tracking.
let response = client
.get_sandbox(with_owner_hint(GetSandboxRequest {
name: name.to_string(),
workspace_scope: Some(openshell_core::proto::workspace_selector(
(workspace).to_string(),
)),
}))
.await
.into_diagnostic()?;
let replica = owner_replica(&response);
let sandbox = response
.into_inner()
.sandbox
.ok_or_else(|| miette::miette!("sandbox not found"))?;
let relay_registration_deadline =
terminal_relay_registration_timeout.map(|timeout| tokio::time::Instant::now() + timeout);
let response = loop {
match client
.create_ssh_session(CreateSshSessionRequest {
sandbox: name.to_string(),
workspace_scope: Some(openshell_core::proto::workspace_selector(
(workspace).to_string(),
)),
})
.await
{
Ok(response) => break response,
Err(status)
if status.code() == Code::FailedPrecondition
&& relay_registration_deadline
.is_some_and(|deadline| tokio::time::Instant::now() < deadline) =>
{
tokio::time::sleep(TERMINAL_RELAY_REGISTRATION_INTERVAL).await;
}
Err(status) => return Err(status).into_diagnostic(),
}
};
let session = response.into_inner();
validate_ssh_session_response(&session)
.map_err(|err| miette::miette!("gateway returned invalid SSH session response: {err}"))?;
let exe = std::env::current_exe()
.into_diagnostic()
.wrap_err("failed to resolve OpenShell executable")?;
let exe_command = exe.to_string_lossy().into_owned();
// When using Cloudflare bearer auth, the SSH CONNECT must go through the
// external tunnel endpoint (the cluster URL), not the server's internal
// scheme/host/port which may be plaintext HTTP on 127.0.0.1.
let gateway_url = if tls.is_bearer_auth() {
server.trim_end_matches('/').to_string()
} else {
// If the server returned a loopback gateway address, override it with the
// cluster endpoint's host. This handles the case where the server defaults
// to 127.0.0.1 but the cluster is actually running on a remote host.
#[allow(clippy::cast_possible_truncation)]
let gateway_port_u16 = session.gateway_port as u16;
let (gateway_scheme, gateway_host, gateway_port) = resolve_ssh_gateway(
&session.gateway_scheme,
&session.gateway_host,
gateway_port_u16,
server,
);
format_gateway_url(&gateway_scheme, &gateway_host, gateway_port)
};
let gateway_name = tls
.gateway_name()
.ok_or_else(|| miette::miette!("gateway name is required to build SSH proxy command"))?;
let proxy_command = build_proxy_command(
&exe_command,
&gateway_url,
name,
workspace,
&session.token,
gateway_name,
);
Ok(SshSessionConfig {
proxy_command,
sandbox_id: session.sandbox_id.clone(),
sandbox_name: name.to_string(),
gateway_url,
token: session.token,
main_terminal: sandbox.spec.as_ref().is_none_or(|spec| spec.tty),
replica,
})
}
fn ssh_base_command(proxy_command: &str) -> Command {
// An explicit environment override wins over the CLI verbosity default.
let ssh_log_level = std::env::var("OPENSHELL_SSH_LOG_LEVEL").unwrap_or_else(|_| {
DEFAULT_SSH_LOG_LEVEL
.get()
.copied()
.unwrap_or("ERROR")
.to_string()
});
let mut command = Command::new("ssh");
// Nested ProxyCommand processes inherit the effective verbosity too.
command.env("OPENSHELL_SSH_LOG_LEVEL", &ssh_log_level);
command
.args(sandbox_ssh_args(proxy_command, &ssh_log_level))
// Detect a dead relay within ~45s. The relay rides on a TCP connection
// that the client has no way to observe silently dropping (gateway
// restart, supervisor restart, cluster failover), so fall back to
// SSH-level keepalives instead of hanging forever.
.arg("-o")
.arg("ServerAliveInterval=15")
.arg("-o")
.arg("ServerAliveCountMax=3");
command
}
async fn exec_or_wait(command: Command, replace_process: bool) -> Result<i32> {
if replace_process && std::io::stdin().is_terminal() {
#[cfg(unix)]
{
let mut command = command;
let err = command.exec();
return Err(miette::miette!("failed to exec ssh: {err}"));
}
}
// Catch terminal signals in the parent while SSH receives the same signals
// from the foreground process group. Unlike SIG_IGN, caught dispositions
// reset to their defaults when the child execs, so SSH stays interruptible.
#[cfg(unix)]
let mut signals = if !replace_process && std::io::stdin().is_terminal() {
Some(TerminationSignals::new()?)
} else {
None
};
let mut child = TokioCommand::from(command)
.kill_on_drop(true)
.spawn()
.into_diagnostic()?;
#[cfg(unix)]
if let Some(signals) = signals.as_mut() {
loop {
tokio::select! {
status = child.wait() => return Ok(process_exit_code(status.into_diagnostic()?)),
_ = signals.recv() => {}
}
}
}
Ok(process_exit_code(child.wait().await.into_diagnostic()?))
}
fn main_attach_command(session: &SshSessionConfig) -> Command {
let mut command = ssh_base_command(&session.proxy_command);
if session.main_terminal {
command.arg("-tt").arg("-o").arg("RequestTTY=force");
} else {
command.arg("-T");
}
command
.arg("-o")
.arg("SetEnv=TERM=xterm-256color")
.arg("-s")
.arg("sandbox")
.arg("openshell-main")
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit());
command
}
async fn run_main_attach(session: &SshSessionConfig, replace_process: bool) -> Result<i32> {
let command = main_attach_command(session);
exec_or_wait(command, replace_process).await
}
fn process_exit_code(status: ExitStatus) -> i32 {
status.code().unwrap_or(1)
}
#[cfg(unix)]
struct TerminationSignals {
interrupt: tokio::signal::unix::Signal,
quit: tokio::signal::unix::Signal,
terminate: tokio::signal::unix::Signal,
}
#[cfg(unix)]
impl TerminationSignals {
fn new() -> Result<Self> {
use tokio::signal::unix::{SignalKind, signal};
Ok(Self {
interrupt: signal(SignalKind::interrupt()).into_diagnostic()?,
quit: signal(SignalKind::quit()).into_diagnostic()?,
terminate: signal(SignalKind::terminate()).into_diagnostic()?,
})
}
async fn recv(&mut self) -> Signal {
tokio::select! {
_ = self.interrupt.recv() => Signal::SIGINT,
_ = self.quit.recv() => Signal::SIGQUIT,
_ = self.terminate.recv() => Signal::SIGTERM,
}
}
}
struct ConnectCancellation {
#[cfg(unix)]
signals: TerminationSignals,
}
impl ConnectCancellation {
// Unix signal registration is fallible; retain the shared API on Windows.
#[cfg_attr(not(unix), allow(clippy::unnecessary_wraps))]
fn new() -> Result<Self> {
Ok(Self {
#[cfg(unix)]
signals: TerminationSignals::new()?,
})
}
// Waiting consumes mutable signal state only on Unix.
#[cfg_attr(not(unix), allow(clippy::needless_pass_by_ref_mut))]
async fn wait<F, T>(&mut self, future: F) -> std::result::Result<T, i32>
where
F: Future<Output = T>,
{
#[cfg(unix)]
{
tokio::select! {
biased;
result = future => Ok(result),
signal = self.signals.recv() => Err(128 + signal as i32),
}
}
#[cfg(not(unix))]
{
Ok(future.await)
}
}
}
#[cfg(unix)]
fn forward_signal_to_child(child: &Child, signal: Signal) -> Result<()> {
let Some(pid) = child.id() else {
return Ok(());
};
let pid = i32::try_from(pid).into_diagnostic()?;
match nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), signal) {
Ok(()) | Err(nix::errno::Errno::ESRCH) => Ok(()),
Err(error) => Err(error).into_diagnostic(),
}
}
#[cfg(unix)]
async fn terminate_and_reap_child(child: &mut Child, signal: Signal) -> Result<i32> {
forward_signal_to_child(child, signal)?;
if let Ok(status) = tokio::time::timeout(CONNECT_CHILD_TERMINATION_TIMEOUT, child.wait()).await
{
status.into_diagnostic()?;
} else {
child.kill().await.into_diagnostic()?;
child.wait().await.into_diagnostic()?;
}
Ok(128 + signal as i32)
}
// The shared cancellation state is mutated by the Unix signal receiver.
#[cfg_attr(not(unix), allow(clippy::needless_pass_by_ref_mut))]
async fn run_main_attach_supervised(
session: &SshSessionConfig,
cancellation: &mut ConnectCancellation,
) -> Result<i32> {
#[cfg(not(unix))]
let _ = cancellation;
let mut command = TokioCommand::from(main_attach_command(session));
command.kill_on_drop(true);
let mut child = command.spawn().into_diagnostic()?;
#[cfg(unix)]
{
tokio::select! {
biased;
status = child.wait() => Ok(process_exit_code(status.into_diagnostic()?)),
signal = cancellation.signals.recv() => terminate_and_reap_child(&mut child, signal).await,
}
}
#[cfg(not(unix))]
{
Ok(process_exit_code(child.wait().await.into_diagnostic()?))
}
}
async fn await_before_recovery_deadline<F, T>(
deadline: Option<Instant>,
future: F,
) -> std::result::Result<T, ()>
where
F: Future<Output = T>,
{
let Some(deadline) = deadline else {
return Ok(future.await);
};
tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), future)
.await
.map_err(|_| ())
}
fn print_connect_recovery_timeout() {
eprintln!(
"Unable to restore the sandbox connection within {} seconds.",
CONNECT_RECOVERY_TIMEOUT.as_secs()
);
}
fn should_recover_connect(
exit_code: i32,
attached_for: Duration,
recovery_active: bool,
main_has_terminal_result: bool,
) -> bool {
!main_has_terminal_result
&& exit_code == SSH_TRANSPORT_FAILURE_EXIT_CODE
&& (recovery_active || attached_for >= CONNECT_ESTABLISHED_DURATION)
}
async fn canonical_main_has_terminal_result(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
) -> bool {
let probe = async {
let Ok(mut client) = grpc_client(server, tls).await else {
return false;
};
loop {
let Ok(response) = client
.get_sandbox(GetSandboxRequest {
name: name.to_string(),
workspace_scope: Some(openshell_core::proto::workspace_selector(
workspace.to_string(),
)),
})
.await
else {
return false;
};
let Some(sandbox) = response.into_inner().sandbox else {
return false;
};
let phase = SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown);
if sandbox
.status
.as_ref()
.is_some_and(|status| status.exit_code.is_some())
|| matches!(
phase,
SandboxPhase::Completed
| SandboxPhase::Error
| SandboxPhase::Stopping
| SandboxPhase::Stopped
| SandboxPhase::Deleting
)
{
return true;
}
tokio::time::sleep(CONNECT_TERMINAL_STATUS_INTERVAL).await;
}
};
tokio::time::timeout(CONNECT_TERMINAL_STATUS_TIMEOUT, probe)
.await
.unwrap_or(false)
}
fn should_start_connect_recovery_window(attached_for: Duration, recovery_active: bool) -> bool {
!recovery_active || attached_for >= CONNECT_RECOVERY_TIMEOUT
}
fn next_connect_retry_delay(delay: Duration) -> Duration {
std::cmp::min(delay.saturating_mul(2), CONNECT_RETRY_MAX_DELAY)
}
fn connect_error_is_retryable(error: &Report) -> bool {
let message = format!("{error:?}").to_ascii_lowercase();
[
"broken pipe",
"connection aborted",
"connection closed",
"connection refused",
"connection reset",
"deadline exceeded",
"h2 protocol",
"http2",
"reset before headers",
"service is currently unavailable",
"timed out",
"timeout",
"transport error",
"unexpected eof",
"unavailable",
"upstream connect error",
]
.iter()
.any(|needle| message.contains(needle))
}
async fn sandbox_connect_supervised(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<i32> {
let mut recovery_deadline = None;
let mut retry_delay = CONNECT_RETRY_INITIAL_DELAY;
let mut cancellation = ConnectCancellation::new()?;
loop {
if recovery_deadline.is_some_and(|deadline| Instant::now() >= deadline) {
print_connect_recovery_timeout();
return Ok(SSH_TRANSPORT_FAILURE_EXIT_CODE);
}
let session_attempt = match Box::pin(cancellation.wait(await_before_recovery_deadline(
recovery_deadline,
ssh_session_config(server, name, tls, workspace, None),
)))
.await
{
Ok(attempt) => attempt,
Err(exit_code) => return Ok(exit_code),
};
let session = match session_attempt {
Err(()) => {
print_connect_recovery_timeout();
return Ok(SSH_TRANSPORT_FAILURE_EXIT_CODE);
}
Ok(Ok(session)) => session,
Ok(Err(error))
if recovery_deadline.is_some_and(|deadline| Instant::now() < deadline)
&& connect_error_is_retryable(&error) =>
{
tracing::warn!(
sandbox = name,
error = %error,
"failed to create replacement SSH session; retrying"
);
let now = Instant::now();
let delay = recovery_deadline
.map_or(retry_delay, |deadline| retry_delay.min(deadline - now));
if let Err(exit_code) = cancellation.wait(tokio::time::sleep(delay)).await {
return Ok(exit_code);
}
retry_delay = next_connect_retry_delay(retry_delay);
continue;
}
Ok(Err(error)) => return Err(error),
};
let attach_started = Instant::now();
let exit_code = run_main_attach_supervised(&session, &mut cancellation).await?;
let attached_for = attach_started.elapsed();
let recovery_active = recovery_deadline.is_some();
let main_has_terminal_result = if exit_code == SSH_TRANSPORT_FAILURE_EXIT_CODE {
match cancellation
.wait(canonical_main_has_terminal_result(
server, name, tls, workspace,
))
.await
{
Ok(has_result) => has_result,
Err(exit_code) => return Ok(exit_code),
}
} else {
false
};
if !should_recover_connect(
exit_code,
attached_for,
recovery_active,
main_has_terminal_result,
) {
return Ok(exit_code);
}
// Once a replacement attachment survives the full recovery interval,
// treat a later failure as a new interruption. Keeping the existing
// deadline for shorter attempts prevents SSH connect timeouts from
// extending recovery forever.
if should_start_connect_recovery_window(attached_for, recovery_active) {
recovery_deadline = Some(Instant::now() + CONNECT_RECOVERY_TIMEOUT);
retry_delay = CONNECT_RETRY_INITIAL_DELAY;
eprintln!(
"Connection to sandbox lost; reconnecting. To disconnect, press Ctrl-D or Ctrl-P then Ctrl-Q after reattachment; press Ctrl-C while retrying to cancel."
);
}
let Some(deadline) = recovery_deadline else {
return Ok(exit_code);
};
let now = Instant::now();
if now >= deadline {
print_connect_recovery_timeout();
return Ok(exit_code);
}
if let Err(exit_code) = cancellation
.wait(tokio::time::sleep(std::cmp::min(
retry_delay,
deadline - now,
)))
.await
{
return Ok(exit_code);
}
retry_delay = next_connect_retry_delay(retry_delay);
}
}
async fn sandbox_connect_with_mode(
server: &str,
name: &str,
tls: &TlsOptions,
replace_process: bool,
workspace: &str,
terminal_relay_registration_timeout: Option<Duration>,
) -> Result<i32> {
let session = ssh_session_config(
server,
name,
tls,
workspace,
terminal_relay_registration_timeout,
)
.await?;
run_main_attach(&session, replace_process).await
}
/// Connect to a sandbox via SSH.
pub async fn sandbox_connect(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<i32> {
sandbox_connect_supervised(server, name, tls, workspace).await
}
pub(crate) async fn sandbox_connect_without_exec(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<i32> {
sandbox_connect_with_mode(server, name, tls, false, workspace, None).await
}
pub(crate) async fn sandbox_connect_terminal_main(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<i32> {
sandbox_connect_with_mode(
server,
name,
tls,
false,
workspace,
Some(TERMINAL_RELAY_REGISTRATION_TIMEOUT),
)
.await
}
pub async fn sandbox_connect_editor(
server: &str,
gateway: &str,
name: &str,
editor: Editor,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
let session = ssh_session_config(server, name, tls, workspace, None).await?;
let workspace_root = discover_workspace_root(&session).await?;
let host_alias = host_alias(name, workspace);
install_ssh_config(gateway, name, workspace)?;
launch_editor(editor, &host_alias, &workspace_root)?;
eprintln!(
"{} Opened {} for sandbox {}",
"✓".green().bold(),
editor.label(),
name
);
Ok(())
}
/// Forward a local port to a sandbox via SSH.
///
/// Background mode keeps the spawned `ssh -N` child alive and records that PID
/// for later management via [`stop_forward`] or [`list_forwards`].
pub async fn sandbox_forward(
server: &str,
name: &str,
spec: &ForwardSpec,
background: bool,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
openshell_core::forward::check_port_available(spec)?;
let session = ssh_session_config(server, name, tls, workspace, None).await?;
let mut command = TokioCommand::from(ssh_forward_command(
&session.proxy_command,
&session.sandbox_id,
spec,
));
if background {
command
.kill_on_drop(false)
.stdin(Stdio::null())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit());
} else {
command
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit());
}
let port = spec.port;
if background {
let mut child = command.spawn().into_diagnostic()?;
let Some(pid) = child.id() else {
terminate_owned_forward_child(&mut child).await;
return Err(miette::miette!(
"ssh process did not expose a PID for background tracking"
));
};
if let Err(err) = wait_for_forward_start(&mut child, spec)
.await
.wrap_err("ssh process started but local forward listener was not reachable")
{
terminate_owned_forward_child(&mut child).await;
return Err(err);
}
let tracked = track_background_forward_or_cleanup(
workspace,
name,
port,
pid,
&session.sandbox_id,
&spec.bind_addr,
|| {
let _ = child.start_kill();
},
);
if tracked.is_err() {
let _ = child.wait().await;
}
tracked?;
return Ok(());
}
let status = {
let mut child = command.spawn().into_diagnostic()?;
if let Err(err) = wait_for_forward_start(&mut child, spec).await {
terminate_owned_forward_child(&mut child).await;
return Err(err);
}
eprintln!("{}", foreground_forward_started_message(name, spec));
child.wait().await.into_diagnostic()?
};
if !status.success() {
return Err(miette::miette!("ssh exited with status {status}"));
}
Ok(())
}
/// A forward must stay owned by the process we spawn so it can be reaped or
/// tracked by PID. User SSH config may otherwise fork it into the background.
fn ssh_forward_command(proxy_command: &str, sandbox_id: &str, spec: &ForwardSpec) -> Command {
let mut command = ssh_base_command(proxy_command);
command
.arg("-o")
.arg("ForkAfterAuthentication=no")
.arg("-N")
.arg("-o")
.arg("ExitOnForwardFailure=yes")
.arg("-o")
.arg(format!("SetEnv=OPENSHELL_FORWARD_SANDBOX_ID={sandbox_id}"))
.arg("-L")
.arg(spec.ssh_forward_arg())
.arg("sandbox");
command
}
/// Wait for the local listener, racing the probe against the `ssh` child
/// exiting. An early exit (e.g. `ExitOnForwardFailure=yes` tearing down the
/// session) means forwarding never came up, so it errors instead of waiting
/// out the grace period.
async fn wait_for_forward_start(child: &mut Child, spec: &ForwardSpec) -> Result<()> {
let listener = wait_for_forward_listener(spec, FORWARD_LISTENER_READINESS_TIMEOUT);
tokio::pin!(listener);
tokio::select! {
result = &mut listener => result,
status = child.wait() => {
let status = status.into_diagnostic()?;
if status.success() {
Err(miette::miette!(
"ssh exited before local forward listener opened on {}:{}",
forward_probe_host(spec),
spec.port,
))
} else {
Err(miette::miette!(
"ssh exited with status {status} before local forward listener opened on {}:{}",
forward_probe_host(spec),
spec.port,
))
}
}
}?;
if let Some(status) = child.try_wait().into_diagnostic()? {
return Err(miette::miette!(
"ssh exited with status {status} after local forward listener opened on {}:{}",
forward_probe_host(spec),
spec.port,
));
}
Ok(())
}
/// Poll the local endpoint until a connect succeeds or `wait_for` elapses. The
/// last probe error is folded into the timeout diagnostic, so a failure reports
/// why the listener never opened, not just that it timed out.
async fn wait_for_forward_listener(spec: &ForwardSpec, wait_for: Duration) -> Result<()> {
wait_for_forward_listener_with_probe(spec, wait_for, |spec| async move {
probe_forward_listener(&spec).await
})
.await
}
async fn wait_for_forward_listener_with_probe<F, Fut>(
spec: &ForwardSpec,
wait_for: Duration,
mut probe: F,
) -> Result<()>
where
F: FnMut(ForwardSpec) -> Fut,
Fut: Future<Output = std::result::Result<(), String>>,
{
let deadline = tokio::time::Instant::now() + wait_for;
loop {
let probe_error = match probe(spec.clone()).await {
Ok(()) => return Ok(()),
Err(err) => err,
};
if tokio::time::Instant::now() >= deadline {
return Err(miette::miette!(
"local forward listener did not open on {}:{} within {}ms: last probe failed with {probe_error}",
forward_probe_host(spec),
spec.port,
wait_for.as_millis(),
));
}
tokio::time::sleep(FORWARD_LISTENER_PROBE_INTERVAL).await;
}
}
/// One bounded TCP connect to the forward endpoint. Returns a `String` error
/// rather than a `miette` diagnostic to stay cheap in the poll loop. The
/// connection only proves reachability and is dropped at once; SSH forwards
/// this throwaway connect to the sandbox-side target.
async fn probe_forward_listener(spec: &ForwardSpec) -> std::result::Result<(), String> {
match tokio::time::timeout(
FORWARD_LISTENER_CONNECT_TIMEOUT,
TcpStream::connect((forward_probe_host(spec), spec.port)),
)
.await
{
Ok(Ok(stream)) => {
drop(stream);
Ok(())
}
Ok(Err(err)) => Err(err.to_string()),
Err(_) => Err(format!(
"connect timed out after {}ms",
FORWARD_LISTENER_CONNECT_TIMEOUT.as_millis()
)),
}
}
/// Resolve the bind address to a connectable host. Wildcard binds (`0.0.0.0`,
/// `::`, empty) are "any-address" listeners, not valid connect targets, so they
/// map to the matching loopback. Specific addresses are probed as-is.
fn forward_probe_host(spec: &ForwardSpec) -> &str {
match spec.bind_addr.as_str() {
"" | "0.0.0.0" => "127.0.0.1",
"::" => "::1",
host => host,
}
}
/// Best-effort cleanup for the SSH child this process spawned. `kill` waits for
/// the child to exit, so failed startup cannot leave an unreaped process.
async fn terminate_owned_forward_child(child: &mut Child) {
let _ = child.kill().await;
}
/// Track a verified background forward, cleaning it up if PID-file persistence fails.
fn track_background_forward_or_cleanup(
workspace: &str,
name: &str,
port: u16,
pid: u32,
sandbox_id: &str,
bind_addr: &str,
cleanup: impl FnOnce(),
) -> Result<()> {
if let Err(err) = write_forward_pid(workspace, name, port, pid, sandbox_id, bind_addr) {
cleanup();
return Err(err)
.wrap_err("local forward listener was reachable but tracking the SSH process failed");
}
Ok(())
}
fn foreground_forward_started_message(name: &str, spec: &ForwardSpec) -> String {
format!(
"{} Forwarding port {} to sandbox {name}\n Access at: {}\n Press Ctrl+C to stop\n {}",
"✓".green().bold(),
spec.port,
spec.access_url(),
"Hint: pass --background to start forwarding without blocking your terminal".dimmed(),
)
}
async fn sandbox_exec_with_mode(
server: &str,
name: &str,
command: &[String],
tty: bool,
tls: &TlsOptions,
replace_process: bool,
workspace: &str,
) -> Result<()> {
if command.is_empty() {
return Err(miette::miette!("no command provided"));
}
let session = ssh_session_config(server, name, tls, workspace, None).await?;
let mut ssh = ssh_base_command(&session.proxy_command);
if tty {
ssh.arg("-tt")
.arg("-o")
.arg("RequestTTY=force")
.arg("-o")
.arg("SetEnv=TERM=xterm-256color");
} else {
ssh.arg("-T").arg("-o").arg("RequestTTY=no");
}
let command_str = command
.iter()
.map(|arg| shell_escape(arg))
.collect::<Vec<_>>()
.join(" ");
ssh.arg("sandbox")
.arg(command_str)
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit());
exec_or_wait(ssh, tty && replace_process).await?;
Ok(())
}
/// Execute a command in a sandbox via SSH.
pub async fn sandbox_exec(
server: &str,
name: &str,
command: &[String],
tty: bool,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
sandbox_exec_with_mode(server, name, command, tty, tls, true, workspace).await
}
/// What to pack into the tar archive streamed to the sandbox.
#[derive(Clone)]
enum UploadSource {
/// A single local file or directory. `tar_name` controls the entry name
/// inside the archive (e.g. the target basename for file-to-file uploads).
SinglePath {
local_path: PathBuf,
tar_name: std::ffi::OsString,
},
/// A set of files relative to a base directory (git-filtered uploads).
FileList {
base_dir: PathBuf,
files: Vec<String>,
archive_prefix: Option<PathBuf>,
},
}
fn write_upload_archive<W: Write>(writer: W, source: UploadSource) -> Result<()> {
let mut archive = tar::Builder::new(writer);
match source {
UploadSource::SinglePath {
local_path,
tar_name,
} => {
append_upload_path(&mut archive, &local_path, Path::new(&tar_name), false)?;
}
UploadSource::FileList {
base_dir,
files,
archive_prefix,
} => {
for file in &files {
let full_path = base_dir.join(file);
let archive_path = archive_prefix
.as_ref()
.map_or_else(|| PathBuf::from(file), |prefix| prefix.join(file));
append_upload_path(&mut archive, &full_path, &archive_path, true)?;
}
}
}
archive.finish().into_diagnostic()?;
Ok(())
}
fn append_upload_path<W: Write>(
archive: &mut tar::Builder<W>,
local_path: &Path,
archive_path: &Path,
skip_missing: bool,
) -> Result<()> {
let metadata = match fs::symlink_metadata(local_path) {
Ok(metadata) => metadata,
Err(err) if skip_missing && err.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(err) => {
return Err(err)
.into_diagnostic()
.wrap_err_with(|| format!("failed to stat {}", local_path.display()));
}
};
let file_type = metadata.file_type();
if file_type.is_file() {
archive
.append_path_with_name(local_path, archive_path)
.into_diagnostic()
.wrap_err_with(|| format!("failed to add {} to tar archive", archive_path.display()))?;
return Ok(());
}
if file_type.is_dir() {
let dir_archive_path = upload_archive_dir_entry_path(archive_path);
archive
.append_dir(&dir_archive_path, local_path)
.into_diagnostic()
.wrap_err_with(|| {
format!(
"failed to add directory {} to tar archive",
archive_path.display()
)
})?;
append_upload_dir_contents(archive, local_path, archive_path)?;
return Ok(());
}
if file_type.is_symlink() {
append_upload_symlink(archive, local_path, archive_path, &metadata)?;
return Ok(());
}
Err(miette::miette!(
"unsupported file type for upload: {}",
local_path.display()
))
}
fn upload_archive_dir_entry_path(archive_path: &Path) -> PathBuf {
let mut path = archive_path.as_os_str().to_os_string();
path.push("/");
PathBuf::from(path)
}
fn append_upload_dir_contents<W: Write>(
archive: &mut tar::Builder<W>,
local_path: &Path,
archive_path: &Path,
) -> Result<()> {
let mut entries = fs::read_dir(local_path)
.into_diagnostic()
.wrap_err_with(|| format!("failed to read directory {}", local_path.display()))?
.collect::<std::io::Result<Vec<_>>>()
.into_diagnostic()
.wrap_err_with(|| format!("failed to read directory {}", local_path.display()))?;
entries.sort_by_key(fs::DirEntry::file_name);
for entry in entries {
let child_local_path = entry.path();
let child_archive_path = archive_path.join(entry.file_name());
append_upload_path(archive, &child_local_path, &child_archive_path, false)?;
}
Ok(())
}
fn append_upload_symlink<W: Write>(
archive: &mut tar::Builder<W>,
local_path: &Path,
archive_path: &Path,
metadata: &fs::Metadata,
) -> Result<()> {
let target = fs::read_link(local_path)
.into_diagnostic()
.wrap_err_with(|| format!("failed to read symlink {}", local_path.display()))?;
let mut header = tar::Header::new_gnu();
header.set_metadata(metadata);
header.set_entry_type(tar::EntryType::Symlink);
header.set_size(0);
header.set_cksum();
archive
.append_link(&mut header, archive_path, target)
.into_diagnostic()
.wrap_err_with(|| {
format!(
"failed to add symlink {} to tar archive",
archive_path.display()
)
})?;
Ok(())
}
fn local_upload_path_is_file_like(path: &Path) -> bool {
fs::symlink_metadata(path).is_ok_and(|metadata| {
let file_type = metadata.file_type();
file_type.is_file() || file_type.is_symlink()
})
}
#[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),
)
}
fn parse_upload_destination_kind(output: &str) -> Result<UploadDestinationKind> {
match output {
"directory" => Ok(UploadDestinationKind::Directory),
"file" => Ok(UploadDestinationKind::File),
"missing" => Ok(UploadDestinationKind::Missing),
_ => Err(miette::miette!(
"unexpected upload destination probe output: {output:?}"
)),
}
}
/// Existing directories receive the source basename. Everything else keeps
/// filename-renaming semantics, including dangling destination symlinks.
fn resolve_file_upload_destination<'a>(
path: &'a str,
source_basename: &std::ffi::OsStr,
kind: UploadDestinationKind,
) -> (&'a str, std::ffi::OsString) {
if kind == UploadDestinationKind::Directory {
(path, source_basename.to_os_string())
} else {
let (parent, basename) = split_sandbox_path(path);
(parent, basename.into())
}
}
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,
)
}
/// Stream an archive into the sandbox and return its final destination.
///
/// Single files and symlinks use the remote destination type to choose between
/// placing the source inside a directory and renaming it. File-list uploads
/// retain their directory-only placement and return that directory.
async fn ssh_tar_upload(
server: &str,
name: &str,
dest: Option<&str>,
mut source: UploadSource,
tls: &TlsOptions,
workspace: &str,
) -> Result<String> {
let session = ssh_session_config(server, name, tls, workspace, None).await?;
let mut dest_dir = dest.unwrap_or(".");
let expected_destination = if let Some(path) = dest
&& let UploadSource::SinglePath {
local_path,
tar_name,
} = &mut source
&& local_upload_path_is_file_like(local_path)
&& file_upload_destination_needs_probe(path)
{
let output = ssh_run_capture_stdout(&session, &build_upload_destination_probe_cmd(path))
.await
.wrap_err_with(|| format!("failed to inspect upload destination '{path}'"))?;
let kind = parse_upload_destination_kind(&output)?;
let (directory, basename) = resolve_file_upload_destination(path, tar_name, kind);
dest_dir = directory;
*tar_name = basename;
Some((path, kind))
} else {
None
};
let final_path = match &source {
UploadSource::SinglePath { tar_name, .. } => {
format!(
"{}/{}",
dest_dir.trim_end_matches('/'),
tar_name.to_string_lossy()
)
}
UploadSource::FileList { .. } => dest_dir.to_string(),
};
let mut ssh = ssh_base_command(&session.proxy_command);
ssh.arg("-T")
.arg("-o")
.arg("RequestTTY=no")
.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)
}
/// Split a sandbox path into (`parent_directory`, basename).
///
/// Examples:
/// `"/sandbox/.bashrc"` -> `("/sandbox", ".bashrc")`
/// `"/sandbox/sub/file"` -> `("/sandbox/sub", "file")`
/// `"file.txt"` -> `(".", "file.txt")`
fn split_sandbox_path(path: &str) -> (&str, &str) {
match path.rfind('/') {
Some(0) => ("/", &path[1..]),
Some(pos) => (&path[..pos], &path[pos + 1..]),
None => (".", path),
}
}
/// Lexically clean a POSIX-style absolute path by resolving `.` and `..`
/// components, collapsing repeated separators, and stripping any trailing
/// slash. Returns `None` if the input is empty or relative — the caller is
/// expected to reject those before reaching this helper.
///
/// This is *lexical* only: it does not consult the filesystem and so cannot
/// follow symlinks. That trade-off is intentional — the function is used
/// client-side to refuse obvious path-traversal attempts before issuing the
/// SSH command. Symlink-based escapes inside the sandbox must be addressed
/// server-side.
fn lexical_clean_absolute_path(path: &str) -> Option<String> {
if !path.starts_with('/') {
return None;
}
let mut stack: Vec<&str> = Vec::new();
for component in path.split('/') {
match component {
"" | "." => {}
".." => {
stack.pop();
}
other => stack.push(other),
}
}
if stack.is_empty() {
return Some("/".to_string());
}
let mut out = String::with_capacity(path.len());
for component in stack {
out.push('/');
out.push_str(component);
}
Some(out)
}
/// Resolve a sandbox-side source path passed to `sandbox download` under the
/// sandbox writable root.
///
/// Returns the cleaned, traversal-resolved path on success. Refuses any
/// path that lexically escapes the discovered workspace root with a user-facing
/// error. Relative paths are interpreted from the workspace root.
///
/// This is a lexical guard only — it does not follow symlinks. Call
/// `resolve_sandbox_source_path` after this on any path that will be passed
/// to a subsequent SSH I/O operation, so a workspace symlink to `/etc` cannot
/// leak files outside the workspace.
fn validate_sandbox_source_path(workspace_root: &str, path: &str) -> Result<String> {
if path.is_empty() {
return Err(miette::miette!("sandbox source path is empty"));
}
let candidate = if path.starts_with('/') {
path.to_string()
} else {
format!("{workspace_root}/{path}")
};
let cleaned = lexical_clean_absolute_path(&candidate)
.ok_or_else(|| miette::miette!("sandbox source path is invalid (got '{path}')"))?;
if !driver_mounts::path_is_or_under(Path::new(&cleaned), Path::new(workspace_root)) {
return Err(miette::miette!(
"sandbox source path '{path}' is outside the sandbox workspace ({workspace_root})"
));
}
Ok(cleaned)
}
/// Discover the workspace root and resolve every symlink in `sandbox_path` in
/// one SSH probe, then refuse the result if it lands outside the workspace.
///
/// The lexical guard in `validate_sandbox_source_path` cannot see symlinks; a
/// workspace path through `etc-link -> /etc` clears the lexical check but
/// would still leak `/etc/passwd` once `tar -C` follows the link. Resolving
/// symlinks on the remote side and re-validating closes that gap. The returned
/// fully-resolved path is what the caller should hand to probe and tar
/// invocations. Combining discovery and resolution also keeps downloads within
/// the gateway's three-connection limit: this probe, the type probe, and tar.
async fn resolve_sandbox_source_path(
session: &SshSessionConfig,
sandbox_path: &str,
) -> Result<String> {
let resolve_cmd = format!(
"pwd -P && realpath -e -- {path}",
path = shell_escape(sandbox_path)
);
let output = ssh_run_capture_stdout(session, &resolve_cmd)
.await
.wrap_err_with(|| format!("failed to resolve sandbox source path '{sandbox_path}'"))?;
let (workspace_root, resolved) = output.split_once('\n').ok_or_else(|| {
miette::miette!("unexpected response while resolving sandbox source path '{sandbox_path}'")
})?;
if resolved.contains('\n') {
return Err(miette::miette!(
"unexpected response while resolving sandbox source path '{sandbox_path}'"
));
}
let workspace_root = validate_discovered_workspace_root(workspace_root)?;
validate_sandbox_source_path(&workspace_root, sandbox_path)?;
if resolved.is_empty() {
return Err(miette::miette!(
"sandbox source path '{sandbox_path}' does not exist"
));
}
if !driver_mounts::path_is_or_under(Path::new(resolved), Path::new(&workspace_root)) {
return Err(miette::miette!(
"sandbox source path '{sandbox_path}' resolves to '{resolved}', outside the sandbox workspace ({workspace_root})"
));
}
Ok(resolved.to_string())
}
/// Resolve the host-side target path for a downloaded *file*, following
/// `cp`-style semantics.
///
/// - If `dest_str` ends with `/` or already exists as a directory, the file is
/// placed inside it as `<dest>/<source_basename>`.
/// - Otherwise `dest_str` is treated as the exact file path to write.
///
/// `dest_exists_as_dir` is taken as a parameter (rather than queried inside)
/// so this function stays pure and unit-testable; the caller performs the
/// filesystem check.
fn resolve_file_download_target(
dest_str: &str,
source_basename: &str,
dest_exists_as_dir: bool,
) -> PathBuf {
let trailing_slash = dest_str.ends_with('/');
let dest_path = Path::new(dest_str);
if trailing_slash || dest_exists_as_dir {
dest_path.join(source_basename)
} else {
dest_path.to_path_buf()
}
}
/// Push a list of files from a local directory into a sandbox using tar-over-SSH.
///
/// Files are streamed as a tar archive to `ssh ... tar xf - -C <dest>` on
/// the sandbox side. When `dest` is `None`, files are uploaded to the
/// SSH session's working directory.
#[allow(clippy::too_many_arguments)]
pub async fn sandbox_sync_up_files(
server: &str,
name: &str,
base_dir: &Path,
files: &[String],
local_path: &Path,
dest: Option<&str>,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
if files.is_empty() {
return Ok(());
}
let source = UploadSource::FileList {
base_dir: base_dir.to_path_buf(),
files: files.to_vec(),
archive_prefix: file_list_archive_prefix(local_path),
};
retry_sandbox_sync("upload", || {
let source = source.clone();
async move { ssh_tar_upload(server, name, dest, source, tls, workspace).await }
})
.await
.map(|_| ())
}
/// Push a local path (file or directory) into a sandbox using tar-over-SSH.
///
/// When `sandbox_path` is `None`, files are uploaded to the SSH session's
/// working directory. A single file or symlink goes inside an existing remote
/// directory, even without a trailing slash. Other nested no-slash destinations
/// rename the source. Returns the final path after a successful upload.
pub async fn sandbox_sync_up(
server: &str,
name: &str,
local_path: &Path,
sandbox_path: Option<&str>,
tls: &TlsOptions,
workspace: &str,
) -> Result<String> {
let local_path_is_file_like = local_upload_path_is_file_like(local_path);
let tar_name = if local_path_is_file_like {
local_path
.file_name()
.ok_or_else(|| miette::miette!("path has no file name"))?
.to_os_string()
} else {
// For directories, wrap contents under the source basename so uploads
// land at `<dest>/<dirname>/...` — matches `scp -r` and `cp -r`. Falls
// back to "." for paths with no meaningful basename (`.`, `/`), which
// preserves the legacy flatten behavior in those edge cases.
directory_upload_prefix(local_path)
};
let source = UploadSource::SinglePath {
local_path: local_path.to_path_buf(),
tar_name,
};
retry_sandbox_sync("upload", || {
let source = source.clone();
async move { ssh_tar_upload(server, name, sandbox_path, source, tls, workspace).await }
})
.await
}
/// Compute the tar entry prefix for a directory upload.
///
/// Returns the directory's basename for any path with a meaningful basename;
/// callers extracting at `<dest>` will see contents wrapped under
/// `<dest>/<basename>/...`. Returns `"."` for paths without a basename
/// (e.g. `.` or `/`), which produces flat extraction at `<dest>`.
fn directory_upload_prefix(local_path: &Path) -> std::ffi::OsString {
local_path
.file_name()
.map_or_else(|| ".".into(), std::ffi::OsStr::to_os_string)
}
fn file_list_archive_prefix(local_path: &Path) -> Option<PathBuf> {
if !local_path.is_dir() {
return None;
}
let prefix = directory_upload_prefix(local_path);
if prefix == "." {
None
} else {
Some(PathBuf::from(prefix))
}
}
/// Run a small command on the sandbox over SSH and capture its stdout.
///
/// Used by the download flow to probe whether the source path is a regular
/// file or a directory before streaming the tar archive. Stderr is inherited
/// so the user still sees any diagnostic output from ssh itself.
async fn ssh_run_capture_stdout(session: &SshSessionConfig, command: &str) -> Result<String> {
let mut ssh = ssh_base_command(&session.proxy_command);
ssh.arg("-T")
.arg("-o")
.arg("RequestTTY=no")
.arg("sandbox")
.arg(command)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::inherit());
let output = tokio::task::spawn_blocking(move || ssh.output())
.await
.into_diagnostic()?
.into_diagnostic()?;
if !output.status.success() {
return Err(miette::miette!(
"ssh probe exited with status {}",
output.status
));
}
decode_ssh_probe_stdout(output.stdout)
}
fn decode_ssh_probe_stdout(stdout: Vec<u8>) -> Result<String> {
let stdout = String::from_utf8(stdout)
.map_err(|error| miette::miette!("ssh probe returned non-UTF-8 output: {error}"))?;
let stdout = stdout.strip_suffix('\n').unwrap_or(&stdout);
let stdout = stdout.strip_suffix('\r').unwrap_or(stdout);
Ok(stdout.to_string())
}
fn validate_discovered_workspace_root(root: &str) -> Result<String> {
let cleaned = lexical_clean_absolute_path(root)
.ok_or_else(|| miette::miette!("remote workspace must be an absolute path"))?;
if cleaned == "/" {
return Err(miette::miette!(
"remote workspace resolved to the container root"
));
}
if cleaned != root {
return Err(miette::miette!(
"remote workspace '{root}' is not a canonical absolute path"
));
}
Ok(cleaned)
}
async fn discover_workspace_root(session: &SshSessionConfig) -> Result<String> {
let root = ssh_run_capture_stdout(session, "pwd -P")
.await
.wrap_err("failed to discover remote workspace")?;
validate_discovered_workspace_root(&root)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum SandboxSourceKind {
File,
Directory,
}
/// Probe the sandbox-side source path. The path is assumed to have already
/// been validated by `validate_sandbox_source_path`.
async fn probe_sandbox_source_kind(
session: &SshSessionConfig,
sandbox_path: &str,
) -> Result<SandboxSourceKind> {
let probe_cmd = format!(
"if [ -d {path} ]; then printf dir; elif [ -e {path} ]; then printf file; else printf missing; fi",
path = shell_escape(sandbox_path),
);
let kind = ssh_run_capture_stdout(session, &probe_cmd).await?;
match kind.as_str() {
"dir" => Ok(SandboxSourceKind::Directory),
"file" => Ok(SandboxSourceKind::File),
"missing" => Err(miette::miette!(
"sandbox source path '{sandbox_path}' does not exist"
)),
other => Err(miette::miette!(
"unexpected probe output for sandbox source path '{sandbox_path}': '{other}'"
)),
}
}
/// Pull a path from a sandbox to a local destination using tar-over-SSH.
///
/// Follows `cp`-style semantics for the destination:
///
/// - If the source is a single file:
/// - When `dest` ends with `/` or already exists as a directory on the host,
/// the file lands at `<dest>/<source-basename>`.
/// - Otherwise `dest` is taken to be the exact file path to write.
/// - If the source is a directory, its contents are extracted into `dest`
/// (creating `dest` if it does not yet exist). This preserves prior
/// behaviour for the directory-source case.
///
/// The sandbox source path is also subjected to a workspace-boundary check
/// before any file probe or archive command is issued; paths that resolve
/// outside the discovered workspace root are refused.
pub async fn sandbox_sync_down(
server: &str,
name: &str,
sandbox_path: &str,
dest: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
retry_sandbox_sync("download", || async {
sandbox_sync_down_once(server, name, sandbox_path, dest, tls, workspace).await
})
.await
}
async fn sandbox_sync_down_once(
server: &str,
name: &str,
sandbox_path: &str,
dest: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
let session = ssh_session_config(server, name, tls, workspace, None).await?;
let sandbox_path = resolve_sandbox_source_path(&session, sandbox_path).await?;
let kind = probe_sandbox_source_kind(&session, &sandbox_path).await?;
match kind {
SandboxSourceKind::File => sandbox_sync_down_file(&session, &sandbox_path, dest).await,
SandboxSourceKind::Directory => {
sandbox_sync_down_directory(&session, &sandbox_path, dest).await
}
}
}
async fn retry_sandbox_sync<F, Fut, T>(operation: &str, mut run: F) -> Result<T>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T>>,
{
let mut attempt = 1;
loop {
match run().await {
Ok(value) => return Ok(value),
Err(error) if attempt < SYNC_RETRY_ATTEMPTS && sync_error_is_retryable(&error) => {
tracing::warn!(
operation,
attempt,
max_attempts = SYNC_RETRY_ATTEMPTS,
error = %error,
"sandbox sync operation failed; retrying"
);
tokio::time::sleep(SYNC_RETRY_DELAY).await;
attempt += 1;
}
Err(error) => return Err(error),
}
}
}
fn sync_error_is_retryable(error: &Report) -> bool {
let message = format!("{error:?}").to_ascii_lowercase();
[
"broken pipe",
"connection",
"early eof",
"http2",
"h2 protocol",
"reset before headers",
"service is currently unavailable",
"transport error",
"unexpected eof",
"unavailable",
"upstream connect error",
"ssh probe exited with status exit status: 255",
"ssh tar create exited",
"ssh tar extract exited",
"failed to extract tar archive from sandbox",
]
.iter()
.any(|needle| message.contains(needle))
}
/// Stream a tar archive from the sandbox and extract it into a fresh
/// destination directory. The source is always wrapped on the sandbox side so
/// the host can pick a basename when needed.
async fn stream_sandbox_tar(
session: &SshSessionConfig,
tar_cmd: String,
extract_into: &Path,
) -> Result<()> {
let mut ssh = ssh_base_command(&session.proxy_command);
ssh.arg("-T")
.arg("-o")
.arg("RequestTTY=no")
.arg("sandbox")
.arg(tar_cmd)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::inherit());
let mut child = ssh.spawn().into_diagnostic()?;
let stdout = child
.stdout
.take()
.ok_or_else(|| miette::miette!("failed to open stdout for ssh process"))?;
let extract_into = extract_into.to_path_buf();
tokio::task::spawn_blocking(move || -> Result<()> {
let mut archive = tar::Archive::new(stdout);
archive
.unpack(&extract_into)
.into_diagnostic()
.wrap_err("failed to extract tar archive from sandbox")?;
Ok(())
})
.await
.into_diagnostic()??;
let status = tokio::task::spawn_blocking(move || child.wait())
.await
.into_diagnostic()?
.into_diagnostic()?;
if !status.success() {
return Err(miette::miette!(
"ssh tar create exited with status {status}"
));
}
Ok(())
}
/// Build the `tar cf - -C <parent> -- <basename>` command used to wrap a
/// single sandbox-side file for download.
///
/// The trailing `--` is required: a sandbox-side file whose basename starts
/// with `-` (e.g. `--checkpoint-action=...`) would otherwise be parsed by GNU
/// tar as an option rather than a member to archive.
fn build_single_file_tar_cmd(parent: &str, basename: &str) -> String {
format!(
"tar cf - -C {parent} -- {name}",
parent = shell_escape(parent),
name = shell_escape(basename),
)
}
async fn sandbox_sync_down_file(
session: &SshSessionConfig,
sandbox_path: &str,
dest: &str,
) -> Result<()> {
let (parent, basename) = split_sandbox_path(sandbox_path);
let dest_exists_as_dir = fs::symlink_metadata(Path::new(dest)).is_ok_and(|m| m.is_dir());
let final_path = resolve_file_download_target(dest, basename, dest_exists_as_dir);
let staging_parent = final_path
.parent()
.ok_or_else(|| miette::miette!("destination '{}' has no parent directory", dest))?;
fs::create_dir_all(staging_parent)
.into_diagnostic()
.wrap_err_with(|| {
format!(
"failed to create local destination directory '{}'",
staging_parent.display()
)
})?;
let staging = tempfile::TempDir::new_in(staging_parent)
.into_diagnostic()
.wrap_err("failed to create download staging directory")?;
let tar_cmd = build_single_file_tar_cmd(parent, basename);
stream_sandbox_tar(session, tar_cmd, staging.path()).await?;
place_downloaded_file(staging.path(), basename, &final_path).wrap_err_with(|| {
format!(
"failed to place downloaded file at '{}'",
final_path.display()
)
})?;
Ok(())
}
/// Move a single file extracted by `stream_sandbox_tar` into its final
/// position on the host.
///
/// `staging_dir` must contain a single regular-file entry named
/// `source_basename` (the wrapper produced by `tar cf - -C <parent> <name>`).
/// The entry is renamed onto `final_path`, atomically when `staging_dir` is
/// on the same filesystem. Refuses to overwrite an existing directory at
/// `final_path` to match `cp` behaviour.
fn place_downloaded_file(
staging_dir: &Path,
source_basename: &str,
final_path: &Path,
) -> Result<()> {
let staged_file = staging_dir.join(source_basename);
let staged_meta = fs::symlink_metadata(&staged_file)
.into_diagnostic()
.wrap_err("downloaded archive did not contain the expected entry")?;
if !staged_meta.is_file() {
return Err(miette::miette!(
"downloaded entry '{source_basename}' is not a regular file"
));
}
if let Ok(existing) = fs::symlink_metadata(final_path)
&& existing.is_dir()
{
return Err(miette::miette!(
"cannot overwrite directory '{}' with downloaded file",
final_path.display()
));
}
fs::rename(&staged_file, final_path)
.into_diagnostic()
.wrap_err_with(|| format!("failed to rename into '{}'", final_path.display()))?;
Ok(())
}
async fn sandbox_sync_down_directory(
session: &SshSessionConfig,
sandbox_path: &str,
dest: &str,
) -> Result<()> {
let dest_path = Path::new(dest);
if let Ok(existing) = fs::symlink_metadata(dest_path)
&& !existing.is_dir()
{
return Err(miette::miette!(
"cannot extract directory '{sandbox_path}' over non-directory destination '{}'",
dest_path.display()
));
}
fs::create_dir_all(dest_path)
.into_diagnostic()
.wrap_err_with(|| {
format!(
"failed to create local destination directory '{}'",
dest_path.display()
)
})?;
let tar_cmd = format!("tar cf - -C {path} .", path = shell_escape(sandbox_path));
stream_sandbox_tar(session, tar_cmd, dest_path).await
}
/// Run the SSH proxy, connecting stdin/stdout to the gateway.
pub async fn sandbox_ssh_proxy(
gateway_url: &str,
sandbox_name: &str,
workspace: &str,
token: &str,
tls: &TlsOptions,
) -> Result<()> {
let server = grpc_server_from_ssh_gateway_url(gateway_url)?;
let mut client = grpc_client(&server, tls).await?;
let replica = client
.get_sandbox(with_owner_hint(GetSandboxRequest {
name: sandbox_name.to_string(),
workspace_scope: Some(openshell_core::proto::workspace_selector(
workspace.to_string(),
)),
}))
.await
.ok()
.and_then(|response| owner_replica(&response));
relay_ssh_proxy(client, sandbox_name, workspace, token, replica.as_ref()).await
}
async fn relay_ssh_proxy(
mut client: GrpcClient,
sandbox_name: &str,
workspace: &str,
token: &str,
replica: Option<&AsciiMetadataValue>,
) -> Result<()> {
let init = TcpForwardFrame {
payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Init(
TcpForwardInit {
sandbox: sandbox_name.to_string(),
workspace: (workspace).to_string(),
service_id: format!("ssh-proxy:{sandbox_name}"),
target: Some(tcp_forward_init::Target::Ssh(SshRelayTarget {})),
authorization_token: token.to_string(),
},
)),
};
let (tx, response) = open_forward_tcp(&mut client, init, replica).await;
let mut response = response.into_diagnostic()?.into_inner();
let stdout = tokio::io::stdout();
// Tokio stdin uses an uncancellable read on the runtime's blocking pool.
// If the relay closes while SSH still holds the pipe open, runtime shutdown
// would wait forever for that read. A dedicated thread can be left behind
// when this ProxyCommand process exits without holding the runtime alive.
let (stdin_tx, mut stdin_rx) = tokio::sync::mpsc::channel(8);
std::thread::Builder::new()
.name("ssh-proxy-stdin".into())
.spawn(move || {
use std::io::Read as _;
let mut stdin = std::io::stdin().lock();
let mut buf = vec![0u8; 64 * 1024];
loop {
match stdin.read(&mut buf) {
Ok(0) | Err(_) => break,
Ok(n) => {
if stdin_tx.blocking_send(buf[..n].to_vec()).is_err() {
break;
}
}
}
}
})
.into_diagnostic()?;
let to_remote = tokio::spawn(async move {
while let Some(data) = stdin_rx.recv().await {
if tx
.send(TcpForwardFrame {
payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Data(
data,
)),
})
.await
.is_err()
{
break;
}
}
});
let from_remote = tokio::spawn(async move {
let mut stdout = stdout;
loop {
let frame = match response.message().await {
Ok(Some(frame)) => frame,
Ok(None) => return Ok::<_, Report>(()),
Err(error) => return Err(error).into_diagnostic(),
};
let Some(openshell_core::proto::tcp_forward_frame::Payload::Data(data)) = frame.payload
else {
continue;
};
if data.is_empty() {
continue;
}
stdout.write_all(&data).await.into_diagnostic()?;
stdout.flush().await.into_diagnostic()?;
}
});
let result = from_remote.await;
to_remote.abort();
result.into_diagnostic()?
}
/// Open a `ForwardTcp` stream whose first frame is `init`, routed to `replica`
/// when known. Returns the sender for the frames that follow.
pub(crate) async fn open_forward_tcp(
client: &mut GrpcClient,
init: TcpForwardFrame,
replica: Option<&AsciiMetadataValue>,
) -> (
tokio::sync::mpsc::Sender<TcpForwardFrame>,
std::result::Result<tonic::Response<tonic::Streaming<TcpForwardFrame>>, tonic::Status>,
) {
let (tx, rx) = tokio::sync::mpsc::channel::<TcpForwardFrame>(16);
let _ = tx.try_send(init.clone());
let response = client
.forward_tcp(routed_to(ReceiverStream::new(rx), replica))
.await;
match response {
Err(status) if retry_unrouted(replica, &status) => {
let (tx, rx) = tokio::sync::mpsc::channel::<TcpForwardFrame>(16);
let _ = tx.try_send(init);
let response = client.forward_tcp(ReceiverStream::new(rx)).await;
(tx, response)
}
response => (tx, response),
}
}
fn grpc_server_from_ssh_gateway_url(gateway_url: &str) -> Result<String> {
let url: url::Url = gateway_url
.parse()
.into_diagnostic()
.wrap_err("invalid gateway URL")?;
let scheme = url.scheme();
let gateway_host = url
.host_str()
.ok_or_else(|| miette::miette!("gateway URL missing host"))?;
let gateway_port = url
.port_or_known_default()
.ok_or_else(|| miette::miette!("gateway URL missing port"))?;
Ok(format_gateway_url(scheme, gateway_host, gateway_port))
}
/// Run the SSH proxy in "name mode": create a session on the fly, then proxy.
///
/// This is equivalent to [`sandbox_ssh_proxy`] but accepts a cluster endpoint
/// and sandbox name instead of pre-created gateway/token credentials. It is
/// suitable for use as an SSH `ProxyCommand` in `~/.ssh/config` because it
/// creates a fresh session on every invocation.
pub async fn sandbox_ssh_proxy_by_name(
server: &str,
name: &str,
tls: &TlsOptions,
workspace: &str,
) -> Result<()> {
let session = ssh_session_config(server, name, tls, workspace, None).await?;
let server = grpc_server_from_ssh_gateway_url(&session.gateway_url)?;
let client = grpc_client(&server, tls).await?;
relay_ssh_proxy(
client,
&session.sandbox_name,
workspace,
&session.token,
session.replica.as_ref(),
)
.await
}
fn host_alias(name: &str, workspace: &str) -> String {
format!("openshell-{name}.{workspace}")
}
fn render_ssh_config(gateway: &str, name: &str, workspace: &str) -> String {
let exe = std::env::current_exe().expect("failed to resolve OpenShell executable");
let exe = shell_escape(&exe.to_string_lossy());
let proxy_cmd = format!(
"{exe} ssh-proxy --gateway-name {} --name {} --workspace {}",
shell_escape(gateway),
shell_escape(name),
shell_escape(workspace),
);
let host_alias = host_alias(name, workspace);
format!(
"Host {host_alias}\n User sandbox\n StrictHostKeyChecking no\n UserKnownHostsFile /dev/null\n GlobalKnownHostsFile /dev/null\n LogLevel ERROR\n ServerAliveInterval 15\n ServerAliveCountMax 3\n ProxyCommand {proxy_cmd}\n"
)
}
fn openshell_ssh_config_path() -> Result<PathBuf> {
Ok(openshell_core::paths::xdg_config_dir()?
.join("openshell")
.join("ssh_config"))
}
fn user_ssh_config_path() -> Result<PathBuf> {
let home = std::env::var("HOME")
.into_diagnostic()
.wrap_err("HOME is not set")?;
Ok(PathBuf::from(home).join(".ssh").join("config"))
}
fn render_include_line(path: &Path) -> String {
format!("Include \"{}\"", path.display())
}
fn ssh_config_includes_path(contents: &str, path: &Path) -> bool {
let quoted = format!("\"{}\"", path.display());
let plain = path.display().to_string();
contents.lines().any(|line| {
let trimmed = line.trim();
if !trimmed.starts_with("Include ") {
return false;
}
trimmed["Include ".len()..]
.split_whitespace()
.any(|token| token == quoted || token == plain)
})
}
fn ensure_openshell_include(main_config: &Path, managed_config: &Path) -> Result<()> {
if let Some(parent) = main_config.parent() {
fs::create_dir_all(parent)
.into_diagnostic()
.wrap_err("failed to create ~/.ssh directory")?;
}
let include_line = render_include_line(managed_config);
let contents = fs::read_to_string(main_config).unwrap_or_default();
let mut lines: Vec<&str> = contents.lines().collect();
lines.retain(|line| !ssh_config_includes_path(line, managed_config));
let insert_at = lines
.iter()
.position(|line| {
let trimmed = line.trim_start();
trimmed.starts_with("Host ") || trimmed.starts_with("Match ")
})
.unwrap_or(lines.len());
let mut out = Vec::new();
out.extend_from_slice(&lines[..insert_at]);
if !out.is_empty() && !out.last().is_some_and(|line| line.is_empty()) {
out.push("");
}
out.push(&include_line);
if insert_at < lines.len() && !lines[insert_at].is_empty() {
out.push("");
}
out.extend_from_slice(&lines[insert_at..]);
let mut rendered = out.join("\n");
if !rendered.is_empty() {
rendered.push('\n');
}
fs::write(main_config, rendered)
.into_diagnostic()
.wrap_err("failed to update ~/.ssh/config")?;
Ok(())
}
fn host_line_matches(line: &str, alias: &str) -> bool {
let trimmed = line.trim_start();
if !trimmed.starts_with("Host ") {
return false;
}
trimmed["Host ".len()..]
.split_whitespace()
.any(|token| token == alias)
}
fn upsert_host_block(contents: &str, alias: &str, block: &str) -> String {
let lines: Vec<&str> = contents.lines().collect();
let start = lines.iter().position(|line| host_line_matches(line, alias));
let mut out = Vec::new();
if let Some(start) = start {
let end = lines
.iter()
.enumerate()
.skip(start + 1)
.find(|(_, line)| line.trim_start().starts_with("Host "))
.map_or(lines.len(), |(idx, _)| idx);
out.extend_from_slice(&lines[..start]);
if !out.is_empty() && !out.last().is_some_and(|line| line.is_empty()) {
out.push("");
}
out.extend(block.lines());
if end < lines.len() && !lines[end..].first().is_some_and(|line| line.is_empty()) {
out.push("");
}
out.extend_from_slice(&lines[end..]);
} else {
out.extend_from_slice(&lines);
if !out.is_empty() && !out.last().is_some_and(|line| line.is_empty()) {
out.push("");
}
out.extend(block.lines());
}
let mut rendered = out.join("\n");
if !rendered.is_empty() {
rendered.push('\n');
}
rendered
}
pub fn install_ssh_config(gateway: &str, name: &str, workspace: &str) -> Result<PathBuf> {
let managed_config = openshell_ssh_config_path()?;
let main_config = user_ssh_config_path()?;
ensure_openshell_include(&main_config, &managed_config)?;
if let Some(parent) = managed_config.parent() {
openshell_core::paths::create_dir_restricted(parent)?;
}
let alias = host_alias(name, workspace);
let block = render_ssh_config(gateway, name, workspace);
let contents = fs::read_to_string(&managed_config).unwrap_or_default();
let updated = upsert_host_block(&contents, &alias, &block);
fs::write(&managed_config, updated)
.into_diagnostic()
.wrap_err("failed to write OpenShell SSH config")?;
Ok(managed_config)
}
fn launch_editor(editor: Editor, host_alias: &str, workspace_root: &str) -> Result<()> {
launch_editor_command(
editor.binary(),
editor.label(),
&Editor::remote_target(host_alias),
workspace_root,
)
}
fn launch_editor_command(
binary: &str,
label: &str,
remote_target: &str,
workspace_root: &str,
) -> Result<()> {
let status = Command::new(binary)
.arg("--remote")
.arg(remote_target)
.arg(workspace_root)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn();
match status {
Ok(_) => Ok(()),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Err(miette::miette!(
"{} is not installed or not on PATH",
binary
)),
Err(err) => Err(err)
.into_diagnostic()
.wrap_err(format!("failed to launch {label}")),
}
}
/// Print an SSH config `Host` block for a sandbox to stdout.
///
/// The output is suitable for appending to `~/.ssh/config` so that tools like
/// `VSCode` Remote-SSH can connect to the sandbox by host alias.
///
/// The `ProxyCommand` uses `--gateway-name` so that `ssh-proxy` resolves the
/// gateway endpoint and TLS certificates from the gateway metadata directory
/// (`~/.config/openshell/gateways/<name>/mtls/`).
pub fn print_ssh_config(gateway: &str, name: &str, workspace: &str) {
print!("{}", render_ssh_config(gateway, name, workspace));
}
#[cfg(test)]
mod tests {
#[test]
fn ssh_verbosity_reaches_nested_proxy_commands_and_respects_override() {
for override_level in [None, Some("INFO")] {
let environment = override_level.map_or_else(
|| Environment::new().remove("OPENSHELL_SSH_LOG_LEVEL"),
|level| Environment::new().set("OPENSHELL_SSH_LOG_LEVEL", level),
);
environment.run(|| {
set_default_ssh_log_level("DEBUG");
let expected = override_level.unwrap_or("DEBUG");
let command = ssh_base_command("openshell ssh-tunnel");
assert!(
command
.get_args()
.any(|arg| arg == std::ffi::OsStr::new(&format!("LogLevel={expected}")))
);
assert!(command.get_envs().any(|(key, value)| {
key == "OPENSHELL_SSH_LOG_LEVEL"
&& value == Some(std::ffi::OsStr::new(expected))
}));
});
}
}
#[cfg(target_os = "linux")]
#[test]
fn interactive_signal_workload_child() {
let Some(marker) = std::env::var_os("OPENSHELL_SIGNAL_WAIT_MARKER") else {
return;
};
// Only publish readiness after exec has reset the parent's caught
// signal dispositions. No external utility filename is assumed.
fs::write(marker, std::process::id().to_string()).unwrap();
std::thread::sleep(Duration::from_secs(30));
}
#[cfg(target_os = "linux")]
#[test]
fn interactive_signal_wait_child() {
let Some(marker) = std::env::var_os("OPENSHELL_SIGNAL_WAIT_MARKER") else {
return;
};
let mut command = Command::new("sh");
command
.args(["-c", "ulimit -c 0; exec \"$@\"", "signal-probe"])
.arg(std::env::current_exe().unwrap())
.args([
"--exact",
"ssh::tests::interactive_signal_workload_child",
"--nocapture",
])
.env("OPENSHELL_SIGNAL_WAIT_MARKER", marker)
.stdin(Stdio::inherit());
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
assert_eq!(runtime.block_on(exec_or_wait(command, false)).unwrap(), 1);
}
#[cfg(target_os = "linux")]
#[test]
fn interactive_wait_survives_terminal_signals_and_reaps_child() {
use nix::sys::signal::killpg;
use nix::unistd::Pid;
struct Probe(std::process::Child);
impl Drop for Probe {
fn drop(&mut self) {
if self.0.try_wait().ok().flatten().is_none() {
let _ = killpg(Pid::from_raw(self.0.id().cast_signed()), Signal::SIGKILL);
}
let _ = self.0.wait();
}
}
for signal in [Signal::SIGINT, Signal::SIGQUIT, Signal::SIGTERM] {
let dir = tempfile::tempdir().unwrap();
let marker = dir.path().join("ready");
let terminal = nix::pty::openpty(None, None).unwrap();
let mut command = Command::new(std::env::current_exe().unwrap());
command
.args([
"--exact",
"ssh::tests::interactive_signal_wait_child",
"--nocapture",
])
.env("OPENSHELL_SIGNAL_WAIT_MARKER", &marker)
.process_group(0)
.stdin(Stdio::from(terminal.slave))
.stdout(Stdio::null())
.stderr(Stdio::null());
let mut probe = Probe(command.spawn().unwrap());
let deadline = Instant::now() + Duration::from_secs(10);
loop {
if let Ok(pid) = fs::read_to_string(&marker)
&& pid.parse::<u32>().is_ok()
{
break;
}
assert!(
probe.0.try_wait().unwrap().is_none(),
"probe exited before ready"
);
assert!(Instant::now() < deadline, "probe did not become ready");
std::thread::sleep(Duration::from_millis(10));
}
killpg(Pid::from_raw(probe.0.id().cast_signed()), signal).unwrap();
loop {
if let Some(status) = probe.0.try_wait().unwrap() {
assert!(status.success(), "parent failed under {signal}: {status}");
break;
}
assert!(
Instant::now() < deadline,
"parent did not reap SSH after {signal}"
);
std::thread::sleep(Duration::from_millis(10));
}
drop(terminal.master);
}
}
use super::*;
use crate::TEST_ENV_LOCK;
use crate::test_utils::Environment;
#[test]
fn upsert_host_block_appends_when_missing() {
let input = "Host existing\n HostName example.com\n";
let block = "Host openshell-demo\n User sandbox\n";
let output = upsert_host_block(input, "openshell-demo", block);
assert!(output.contains("Host existing"));
assert!(output.contains("Host openshell-demo"));
assert_eq!(output.matches("Host openshell-demo").count(), 1);
}
#[test]
fn upsert_host_block_replaces_existing_without_duplicates() {
let input = "Host openshell-demo\n User old\n\nHost other\n HostName other.example\n";
let block = "Host openshell-demo\n User sandbox\n LogLevel ERROR\n";
let output = upsert_host_block(input, "openshell-demo", block);
assert!(!output.contains("User old"));
assert!(output.contains("LogLevel ERROR"));
assert!(output.contains("Host other"));
assert_eq!(output.matches("Host openshell-demo").count(), 1);
}
#[test]
fn sync_error_retry_filter_accepts_transport_failures() {
let error = miette::miette!("transport error: connection reset by peer");
assert!(sync_error_is_retryable(&error));
}
#[test]
fn sync_error_retry_filter_accepts_transient_ssh_probe_failures() {
let error = Err::<(), _>(miette::miette!(
"ssh probe exited with status exit status: 255"
))
.wrap_err("failed to resolve sandbox source path '/sandbox/ha-sync/ha-sync-upload'")
.unwrap_err();
assert!(sync_error_is_retryable(&error));
}
#[test]
fn sync_error_retry_filter_rejects_validation_failures() {
let error = miette::miette!("sandbox source path '/etc/passwd' resolves outside /sandbox");
assert!(!sync_error_is_retryable(&error));
}
#[test]
fn connect_recovery_starts_only_for_established_transport_failures() {
assert!(should_recover_connect(
SSH_TRANSPORT_FAILURE_EXIT_CODE,
CONNECT_ESTABLISHED_DURATION,
false,
false
));
assert!(!should_recover_connect(
SSH_TRANSPORT_FAILURE_EXIT_CODE,
CONNECT_ESTABLISHED_DURATION
.checked_sub(Duration::from_millis(1))
.unwrap(),
false,
false
));
assert!(!should_recover_connect(
0,
CONNECT_ESTABLISHED_DURATION,
false,
false
));
assert!(!should_recover_connect(
SSH_TRANSPORT_FAILURE_EXIT_CODE,
CONNECT_ESTABLISHED_DURATION,
false,
true
));
}
#[test]
fn connect_recovery_continues_after_fast_replacement_failure() {
assert!(should_recover_connect(
SSH_TRANSPORT_FAILURE_EXIT_CODE,
Duration::ZERO,
true,
false
));
assert!(!should_start_connect_recovery_window(
CONNECT_ESTABLISHED_DURATION,
true
));
assert!(should_start_connect_recovery_window(
CONNECT_RECOVERY_TIMEOUT,
true
));
}
#[test]
fn connect_retry_delay_is_capped() {
assert_eq!(
next_connect_retry_delay(CONNECT_RETRY_INITIAL_DELAY),
Duration::from_millis(500)
);
assert_eq!(
next_connect_retry_delay(CONNECT_RETRY_MAX_DELAY),
CONNECT_RETRY_MAX_DELAY
);
}
#[test]
fn connect_retry_filter_rejects_authentication_failures() {
let transient = miette::miette!("transport error: connection reset by peer");
assert!(connect_error_is_retryable(&transient));
let authentication = miette::miette!("status: Unauthenticated");
assert!(!connect_error_is_retryable(&authentication));
let lifecycle = miette::miette!("sandbox is not ready");
assert!(!connect_error_is_retryable(&lifecycle));
}
#[tokio::test]
async fn connect_recovery_deadline_bounds_stalled_session_creation() {
let deadline = Instant::now() + Duration::from_millis(20);
let result = await_before_recovery_deadline(
Some(deadline),
std::future::pending::<std::result::Result<(), Report>>(),
)
.await;
assert!(result.is_err());
assert!(Instant::now() >= deadline);
}
#[test]
fn install_ssh_config_adds_include_once_and_updates_managed_file() {
let home = tempfile::tempdir().unwrap();
let xdg = tempfile::tempdir().unwrap();
Environment::new()
.set("HOME", home.path())
.set("XDG_CONFIG_HOME", xdg.path())
.run(|| {
let home = PathBuf::from(std::env::var_os("HOME").unwrap());
let ssh_dir = home.join(".ssh");
fs::create_dir_all(&ssh_dir).unwrap();
let user_config = ssh_dir.join("config");
fs::write(&user_config, "Host personal\n HostName example.com\n").unwrap();
let managed_path = install_ssh_config("openshell", "demo", "default").unwrap();
install_ssh_config("openshell", "demo", "default").unwrap();
let main_contents = fs::read_to_string(&user_config).unwrap();
assert!(main_contents.contains("Host personal"));
assert_eq!(main_contents.matches("Include ").count(), 1);
assert!(main_contents.contains(&render_include_line(&managed_path)));
let include_idx = main_contents.find("Include ").unwrap();
let host_idx = main_contents.find("Host personal").unwrap();
assert!(include_idx < host_idx);
let managed_contents = fs::read_to_string(&managed_path).unwrap();
assert_eq!(
managed_contents
.matches("Host openshell-demo.default")
.count(),
1
);
assert!(managed_contents.contains("ProxyCommand"));
});
}
#[test]
fn render_ssh_config_includes_workspace_in_proxy_command() {
let config = render_ssh_config("my-gw", "demo", "beta");
assert!(
config.contains("Host openshell-demo.beta"),
"host alias should be workspace-qualified: {config}"
);
assert!(
config.contains("--workspace beta"),
"ProxyCommand should include --workspace: {config}"
);
}
#[test]
fn host_alias_includes_workspace() {
assert_eq!(host_alias("demo", "default"), "openshell-demo.default");
assert_eq!(host_alias("demo", "beta"), "openshell-demo.beta");
}
#[test]
fn launch_editor_returns_friendly_error_when_binary_missing() {
let err = launch_editor_command(
"openshell-test-missing-binary",
"Test Editor",
"ssh-remote+openshell-demo",
"/workspace/project",
)
.unwrap_err();
let text = format!("{err}");
assert!(text.contains("openshell-test-missing-binary is not installed or not on PATH"));
}
#[test]
fn foreground_forward_started_message_includes_port_and_stop_hint() {
let spec = ForwardSpec::new(8080);
let message = foreground_forward_started_message("demo", &spec);
assert!(message.contains("Forwarding port 8080 to sandbox demo"));
assert!(message.contains("Access at: http://127.0.0.1:8080/"));
assert!(message.contains("sandbox demo"));
assert!(message.contains("Press Ctrl+C to stop"));
assert!(message.contains(
"Hint: pass --background to start forwarding without blocking your terminal"
));
}
#[test]
fn foreground_forward_started_message_custom_bind_addr() {
let spec = ForwardSpec::parse("0.0.0.0:3000").unwrap();
let message = foreground_forward_started_message("demo", &spec);
assert!(message.contains("Forwarding port 3000 to sandbox demo"));
assert!(message.contains("Access at: http://localhost:3000/"));
}
#[test]
fn forward_probe_host_uses_connectable_loopback_for_wildcard_binds() {
let ipv4 = ForwardSpec::parse("0.0.0.0:3000").unwrap();
let ipv6 = ForwardSpec::parse(":::3000").unwrap();
let loopback = ForwardSpec::new(3000);
assert_eq!(forward_probe_host(&ipv4), "127.0.0.1");
assert_eq!(forward_probe_host(&ipv6), "::1");
assert_eq!(forward_probe_host(&loopback), "127.0.0.1");
}
/// Resolve `ssh -G` for `command`'s arguments against a user config that
/// enables connection sharing and forking for the host `sandbox`. `extra_args`
/// supplies the host and, for the forking option, `-N`, when `command` lacks them.
#[cfg(unix)]
fn effective_ssh_config(command: &Command, extra_args: &[&str]) -> String {
let _guard = TEST_ENV_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let dir = tempfile::tempdir().unwrap();
let config = dir.path().join("ssh_config");
fs::write(
&config,
format!(
"Host sandbox\n ControlMaster auto\n ControlPath {}/mux-%C\n ControlPersist yes\n ForkAfterAuthentication yes\n",
dir.path().display()
),
)
.unwrap();
let output = Command::new("ssh")
.arg("-F")
.arg(&config)
.arg("-G")
.args(command.get_args())
.args(extra_args)
.output()
.unwrap();
assert!(
output.status.success(),
"ssh -G failed: {}",
String::from_utf8_lossy(&output.stderr)
);
String::from_utf8(output.stdout).unwrap()
}
#[cfg(unix)]
fn assert_connection_sharing_disabled(effective: &str) {
assert!(effective.lines().any(|line| line == "controlmaster false"));
assert!(effective.lines().any(|line| line == "controlpersist no"));
assert!(
!effective
.lines()
.any(|line| { line.starts_with("controlpath ") && line != "controlpath none" })
);
}
#[cfg(unix)]
#[test]
fn ssh_base_command_overrides_user_connection_sharing() {
let effective =
effective_ssh_config(&ssh_base_command("openshell ssh-proxy"), &["-N", "sandbox"]);
assert_connection_sharing_disabled(&effective);
}
#[cfg(unix)]
#[test]
fn forward_ssh_command_overrides_user_multiplexing_and_forking() {
let forward = ssh_forward_command(
"openshell ssh-proxy --sandbox demo",
"sandbox-id",
&ForwardSpec::new(18080),
);
let effective = effective_ssh_config(&forward, &[]);
assert_connection_sharing_disabled(&effective);
assert!(
effective
.lines()
.any(|line| line == "forkafterauthentication no")
);
}
#[tokio::test]
async fn wait_for_forward_listener_accepts_ready_listener() {
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.unwrap();
let port = listener.local_addr().unwrap().port();
let accept = tokio::spawn(async move {
let _ = listener.accept().await;
});
let spec = ForwardSpec::new(port);
wait_for_forward_listener(&spec, Duration::from_secs(1))
.await
.unwrap();
accept.await.unwrap();
}
#[tokio::test]
async fn wait_for_forward_listener_reports_failed_probe() {
let spec = ForwardSpec::new(12345);
let err = wait_for_forward_listener_with_probe(&spec, Duration::ZERO, |_| {
std::future::ready(Err("connection refused".to_string()))
})
.await
.unwrap_err();
let text = format!("{err:?}");
assert!(text.contains("local forward listener did not open"));
assert!(text.contains("connection refused"));
}
#[test]
fn track_background_forward_or_cleanup_runs_cleanup_when_pidfile_write_fails() {
let tmp = tempfile::tempdir().unwrap();
// Make forward PID-file writes fail with ENOTDIR after listener readiness.
let blocking_file = tmp.path().join("not-a-dir");
fs::write(&blocking_file, b"x").unwrap();
Environment::new()
.set("XDG_CONFIG_HOME", &blocking_file)
.run(|| {
let mut cleaned_up = false;
let result = track_background_forward_or_cleanup(
"default",
"demo",
8080,
4242,
"sbx-1",
"127.0.0.1",
|| {
cleaned_up = true;
},
);
assert!(
result.is_err(),
"PID-file write failure must surface as an error"
);
assert!(
cleaned_up,
"the owned SSH child must be cleaned up when tracking fails so no \
reachable-but-untracked forward is left running"
);
});
}
#[test]
fn track_background_forward_or_cleanup_tracks_pid_without_cleanup_on_success() {
let tmp = tempfile::tempdir().unwrap();
Environment::new()
.set("XDG_CONFIG_HOME", tmp.path())
.run(|| {
let mut cleaned_up = false;
let result = track_background_forward_or_cleanup(
"default",
"demo",
8080,
4242,
"sbx-1",
"127.0.0.1",
|| {
cleaned_up = true;
},
);
let pid_file_exists =
openshell_core::forward::forward_pid_path("default", "demo", 8080)
.is_ok_and(|path| path.exists());
assert!(
result.is_ok(),
"a writable PID directory must track successfully"
);
assert!(
pid_file_exists,
"successful tracking must persist a PID file"
);
assert!(
!cleaned_up,
"successful tracking must not terminate the forward process"
);
});
}
#[test]
fn split_sandbox_path_separates_parent_and_basename() {
assert_eq!(
split_sandbox_path("/sandbox/.bashrc"),
("/sandbox", ".bashrc")
);
assert_eq!(
split_sandbox_path("/sandbox/sub/file"),
("/sandbox/sub", "file")
);
assert_eq!(split_sandbox_path("/a/b/c/d.txt"), ("/a/b/c", "d.txt"));
}
#[test]
fn lexical_clean_resolves_dot_and_dotdot_segments() {
assert_eq!(
lexical_clean_absolute_path("/sandbox/./a"),
Some("/sandbox/a".to_string())
);
assert_eq!(
lexical_clean_absolute_path("/sandbox/sub/../a"),
Some("/sandbox/a".to_string())
);
assert_eq!(
lexical_clean_absolute_path("/sandbox/../etc/passwd"),
Some("/etc/passwd".to_string())
);
assert_eq!(
lexical_clean_absolute_path("//sandbox///foo//"),
Some("/sandbox/foo".to_string())
);
assert_eq!(lexical_clean_absolute_path("/"), Some("/".to_string()));
}
#[test]
fn lexical_clean_refuses_relative_paths() {
assert_eq!(lexical_clean_absolute_path(""), None);
assert_eq!(lexical_clean_absolute_path("sandbox/a"), None);
assert_eq!(lexical_clean_absolute_path("./a"), None);
}
#[test]
fn validate_sandbox_source_path_accepts_workspace_paths() {
let workspace_root = "/workspace/project";
assert_eq!(
validate_sandbox_source_path(workspace_root, "/workspace/project/file.txt").unwrap(),
"/workspace/project/file.txt"
);
assert_eq!(
validate_sandbox_source_path(
workspace_root,
"/workspace/project/.agent/workspace/hello.txt"
)
.unwrap(),
"/workspace/project/.agent/workspace/hello.txt"
);
assert_eq!(
validate_sandbox_source_path(workspace_root, "/workspace/project").unwrap(),
"/workspace/project"
);
assert_eq!(
validate_sandbox_source_path(workspace_root, "/workspace/project/").unwrap(),
"/workspace/project"
);
assert_eq!(
validate_sandbox_source_path(workspace_root, "/workspace/project/sub/../file").unwrap(),
"/workspace/project/file"
);
assert_eq!(
validate_sandbox_source_path(workspace_root, "output/file.txt").unwrap(),
"/workspace/project/output/file.txt"
);
assert_eq!(
validate_sandbox_source_path(workspace_root, "./output/../file.txt").unwrap(),
"/workspace/project/file.txt"
);
}
#[test]
fn validate_sandbox_source_path_rejects_traversal_and_escapes() {
let workspace_root = "/workspace/project";
let traversal = validate_sandbox_source_path(workspace_root, "/etc/passwd").unwrap_err();
assert!(
format!("{traversal}").contains("outside the sandbox workspace"),
"unexpected error: {traversal}"
);
let parent_escape =
validate_sandbox_source_path(workspace_root, "/workspace/project/../../etc/passwd")
.unwrap_err();
assert!(
format!("{parent_escape}").contains("outside the sandbox workspace"),
"unexpected error: {parent_escape}"
);
let prefix_only =
validate_sandbox_source_path(workspace_root, "/workspace/projected/secrets")
.unwrap_err();
assert!(
format!("{prefix_only}").contains("outside the sandbox workspace"),
"unexpected error: {prefix_only}"
);
let empty = validate_sandbox_source_path(workspace_root, "").unwrap_err();
assert!(format!("{empty}").contains("empty"));
let relative_escape =
validate_sandbox_source_path(workspace_root, "../../etc/passwd").unwrap_err();
assert!(format!("{relative_escape}").contains("outside the sandbox workspace"));
}
#[test]
fn discovered_workspace_root_must_be_canonical_absolute_non_root() {
assert_eq!(
validate_discovered_workspace_root("/workspace/project").unwrap(),
"/workspace/project"
);
for invalid in ["", "workspace", "/", "/workspace/../etc", "/workspace/"] {
assert!(
validate_discovered_workspace_root(invalid).is_err(),
"expected '{invalid}' to be rejected"
);
}
}
#[test]
fn ssh_probe_output_only_removes_the_protocol_line_ending() {
assert_eq!(
decode_ssh_probe_stdout(b"/workspace/project \n".to_vec()).unwrap(),
"/workspace/project "
);
assert_eq!(
decode_ssh_probe_stdout(b"/workspace/project\r\n".to_vec()).unwrap(),
"/workspace/project"
);
assert!(decode_ssh_probe_stdout(vec![0xff]).is_err());
}
#[test]
fn build_single_file_tar_cmd_inserts_double_dash_before_basename() {
// Without `--`, a basename such as `--checkpoint-action=...` would be
// parsed by GNU tar as an option. Guard the wire format against this
// regression.
let cmd = build_single_file_tar_cmd("/sandbox", "--checkpoint-action=exec=id");
assert!(
cmd.contains(" -- "),
"expected `--` separator in tar command, got: {cmd}"
);
assert!(
cmd.ends_with(&shell_escape("--checkpoint-action=exec=id")),
"expected basename at end of tar command, got: {cmd}"
);
}
#[test]
fn build_single_file_tar_cmd_escapes_parent_and_basename() {
let cmd = build_single_file_tar_cmd("/sandbox/with space", "name with space");
assert!(cmd.contains(" -- "), "missing `--` separator: {cmd}");
assert!(
cmd.contains(&shell_escape("/sandbox/with space")),
"parent not shell-escaped: {cmd}"
);
assert!(
cmd.contains(&shell_escape("name with space")),
"basename not shell-escaped: {cmd}"
);
}
#[test]
fn resolve_file_download_target_writes_to_dest_when_not_a_directory() {
assert_eq!(
resolve_file_download_target("/tmp/out.txt", "hello.txt", false),
PathBuf::from("/tmp/out.txt")
);
}
#[test]
fn resolve_file_download_target_places_inside_existing_directory() {
assert_eq!(
resolve_file_download_target("/tmp", "hello.txt", true),
PathBuf::from("/tmp/hello.txt")
);
}
#[test]
fn resolve_file_download_target_honors_trailing_slash() {
assert_eq!(
resolve_file_download_target("/tmp/newdir/", "hello.txt", false),
PathBuf::from("/tmp/newdir/hello.txt")
);
}
fn build_single_file_archive(entry_path: &str, bytes: &[u8]) -> Vec<u8> {
let mut buf = Vec::new();
{
let mut builder = tar::Builder::new(&mut buf);
let mut header = tar::Header::new_gnu();
header.set_path(entry_path).expect("set tar entry path");
header.set_size(bytes.len() as u64);
header.set_mode(0o644);
header.set_entry_type(tar::EntryType::Regular);
header.set_cksum();
builder.append(&header, bytes).expect("append tar entry");
builder.finish().expect("finish tar archive");
}
buf
}
fn unpack_into(archive_bytes: &[u8], staging: &Path) {
let mut archive = tar::Archive::new(std::io::Cursor::new(archive_bytes));
archive.unpack(staging).expect("unpack archive");
}
#[test]
fn place_downloaded_file_writes_regular_file_at_dest() {
let workdir = tempfile::tempdir().expect("create workdir");
let staging = workdir.path().join("staging");
fs::create_dir_all(&staging).expect("create staging");
let archive = build_single_file_archive("hello.txt", b"trust me");
unpack_into(&archive, &staging);
let dest = workdir.path().join("out.txt");
place_downloaded_file(&staging, "hello.txt", &dest).expect("place file");
let meta = fs::symlink_metadata(&dest).expect("stat dest");
assert!(meta.is_file(), "dest must be a regular file, got {meta:?}");
assert_eq!(fs::read(&dest).expect("read dest"), b"trust me");
}
#[test]
fn place_downloaded_file_refuses_to_clobber_existing_directory() {
let workdir = tempfile::tempdir().expect("create workdir");
let staging = workdir.path().join("staging");
fs::create_dir_all(&staging).expect("create staging");
let archive = build_single_file_archive("hello.txt", b"trust me");
unpack_into(&archive, &staging);
let dest = workdir.path().join("conflict-dir");
fs::create_dir(&dest).expect("create conflict dir");
let err = place_downloaded_file(&staging, "hello.txt", &dest)
.expect_err("expected directory-clobber refusal");
assert!(
format!("{err}").contains("cannot overwrite directory"),
"unexpected error: {err}"
);
assert!(
fs::symlink_metadata(&dest).expect("stat dest").is_dir(),
"dest should remain a directory after refusal"
);
}
#[test]
fn download_full_pipeline_lands_file_at_exact_dest_path() {
let workdir = tempfile::tempdir().expect("create workdir");
let staging_parent = workdir.path();
let archive = build_single_file_archive("hello.txt", b"trust me");
let dest_str = staging_parent.join("out.txt");
let dest_str = dest_str.to_str().unwrap();
let final_path = resolve_file_download_target(dest_str, "hello.txt", false);
assert_eq!(final_path, Path::new(dest_str));
let staging = tempfile::TempDir::new_in(staging_parent).expect("staging dir");
unpack_into(&archive, staging.path());
place_downloaded_file(staging.path(), "hello.txt", &final_path).expect("place");
let meta = fs::symlink_metadata(&final_path).expect("stat final");
assert!(meta.is_file(), "expected regular file, got {meta:?}");
assert_eq!(fs::read(&final_path).expect("read final"), b"trust me");
}
#[test]
fn download_full_pipeline_places_inside_existing_directory_destination() {
let workdir = tempfile::tempdir().expect("create workdir");
let archive = build_single_file_archive("hello.txt", b"trust me");
let dest_dir = workdir.path().join("out-dir");
fs::create_dir(&dest_dir).expect("create dest dir");
let dest_str = dest_dir.to_str().unwrap();
let final_path = resolve_file_download_target(dest_str, "hello.txt", true);
assert_eq!(final_path, dest_dir.join("hello.txt"));
let staging = tempfile::TempDir::new_in(workdir.path()).expect("staging dir");
unpack_into(&archive, staging.path());
place_downloaded_file(staging.path(), "hello.txt", &final_path).expect("place");
let meta = fs::symlink_metadata(&final_path).expect("stat final");
assert!(meta.is_file());
assert_eq!(fs::read(&final_path).expect("read final"), b"trust me");
}
#[cfg(unix)]
fn run_upload_shell(
command: &str,
workdir: &Path,
archive: Option<&[u8]>,
) -> std::process::Output {
let mut child = Command::new("sh")
.args(["-c", command])
.current_dir(workdir)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("start upload shell");
let write_result = child
.stdin
.take()
.map(|mut stdin| stdin.write_all(archive.unwrap_or_default()));
let output = child.wait_with_output().expect("wait for upload shell");
if output.status.success() {
write_result.unwrap().expect("write upload archive");
}
output
}
#[cfg(unix)]
fn prepare_file_upload_command(
local_path: &Path,
dest: &str,
workdir: &Path,
) -> (String, Vec<u8>) {
let basename = local_path.file_name().expect("source basename");
let (directory, tar_name, expected) = if file_upload_destination_needs_probe(dest) {
let probe = run_upload_shell(&build_upload_destination_probe_cmd(dest), workdir, None);
assert!(
probe.status.success(),
"destination probe failed: {}",
String::from_utf8_lossy(&probe.stderr)
);
let kind = parse_upload_destination_kind(
&decode_ssh_probe_stdout(probe.stdout).expect("decode probe"),
)
.expect("destination kind");
let (directory, tar_name) = resolve_file_upload_destination(dest, basename, kind);
(directory, tar_name, Some((dest, kind)))
} else {
(dest, basename.to_os_string(), None)
};
let mut archive = Vec::new();
write_upload_archive(
&mut archive,
UploadSource::SinglePath {
local_path: local_path.to_path_buf(),
tar_name,
},
)
.expect("write upload archive");
(build_upload_extract_cmd(directory, expected), archive)
}
#[cfg(unix)]
fn upload_file_locally(local_path: &Path, dest: &str, workdir: &Path) {
let (command, archive) = prepare_file_upload_command(local_path, dest, workdir);
let output = run_upload_shell(&command, workdir, Some(&archive));
assert!(
output.status.success(),
"upload failed: {}",
String::from_utf8_lossy(&output.stderr)
);
}
#[cfg(unix)]
#[test]
fn upload_into_existing_empty_directory_preserves_directory() {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("config.json");
fs::write(&source, b"configuration").unwrap();
let directory = tmp.path().join("gym");
fs::create_dir(&directory).unwrap();
upload_file_locally(&source, directory.to_str().unwrap(), tmp.path());
assert!(directory.is_dir(), "upload replaced the existing directory");
assert_eq!(
fs::read(directory.join("config.json")).unwrap(),
b"configuration"
);
}
#[cfg(unix)]
#[test]
fn upload_preserves_nonempty_directory_and_directory_symlink() {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("config.json");
fs::write(&source, b"configuration").unwrap();
let directory = tmp.path().join("gym");
fs::create_dir(&directory).unwrap();
fs::write(directory.join("keep.txt"), b"keep").unwrap();
let link = tmp.path().join("linked-gym");
std::os::unix::fs::symlink("gym", &link).unwrap();
for destination in [&directory, &link] {
upload_file_locally(&source, destination.to_str().unwrap(), tmp.path());
assert_eq!(
fs::read(directory.join("config.json")).unwrap(),
b"configuration"
);
assert_eq!(fs::read(directory.join("keep.txt")).unwrap(), b"keep");
}
assert_eq!(fs::read_link(&link).unwrap(), Path::new("gym"));
}
#[cfg(unix)]
#[test]
fn upload_retains_file_renames_overwrites_and_trailing_slash_placement() {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("config.json");
fs::write(&source, b"configuration").unwrap();
fs::write(tmp.path().join("existing.json"), b"old").unwrap();
for dest in ["missing/renamed.json", "existing.json", "bare-name"] {
upload_file_locally(&source, dest, tmp.path());
assert_eq!(fs::read(tmp.path().join(dest)).unwrap(), b"configuration");
}
upload_file_locally(&source, "new-directory/", tmp.path());
assert_eq!(
fs::read(tmp.path().join("new-directory/config.json")).unwrap(),
b"configuration"
);
}
#[cfg(unix)]
#[test]
fn upload_preserves_dangling_source_symlink_and_replaces_dangling_destination() {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("source-link");
std::os::unix::fs::symlink("missing-target", &source).unwrap();
let directory = tmp.path().join("gym");
fs::create_dir(&directory).unwrap();
upload_file_locally(&source, "gym", tmp.path());
assert_eq!(
fs::read_link(directory.join("source-link")).unwrap(),
Path::new("missing-target")
);
let regular_source = tmp.path().join("config.json");
fs::write(&regular_source, b"configuration").unwrap();
let dangling_dest = tmp.path().join("dangling-dest");
std::os::unix::fs::symlink("missing-target", &dangling_dest).unwrap();
upload_file_locally(&regular_source, "dangling-dest", tmp.path());
assert!(fs::symlink_metadata(&dangling_dest).unwrap().is_file());
assert_eq!(fs::read(&dangling_dest).unwrap(), b"configuration");
assert!(!tmp.path().join("missing-target").exists());
}
#[cfg(unix)]
#[test]
fn upload_quotes_shell_metacharacters_and_leading_dashes() {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("config '$(touch source-injected)'\n.json");
fs::write(&source, b"configuration").unwrap();
let dest = "-gym '$(touch destination-injected)'\n";
let directory = tmp.path().join(dest);
fs::create_dir(&directory).unwrap();
upload_file_locally(&source, dest, tmp.path());
assert_eq!(
fs::read(directory.join(source.file_name().unwrap())).unwrap(),
b"configuration"
);
assert!(!tmp.path().join("source-injected").exists());
assert!(!tmp.path().join("destination-injected").exists());
}
#[cfg(unix)]
#[test]
fn upload_probe_rejects_invalid_parents_instead_of_reporting_missing() {
let tmp = tempfile::tempdir().unwrap();
fs::write(tmp.path().join("file-parent"), b"keep").unwrap();
std::os::unix::fs::symlink("missing-parent", tmp.path().join("link-parent")).unwrap();
for dest in [
"file-parent/child",
"file-parent/nested/child",
"link-parent/child",
] {
let output =
run_upload_shell(&build_upload_destination_probe_cmd(dest), tmp.path(), None);
assert!(!output.status.success(), "invalid parent accepted: {dest}");
assert!(
output.stdout.is_empty(),
"probe must not emit missing on failure"
);
assert!(String::from_utf8_lossy(&output.stderr).contains("parent is not a directory"));
}
assert!(parse_upload_destination_kind("").is_err());
assert!(parse_upload_destination_kind("directory\nunexpected").is_err());
}
#[cfg(unix)]
#[test]
fn upload_probe_rejects_unsearchable_parent_instead_of_reporting_missing() {
use std::os::unix::fs::PermissionsExt;
let tmp = tempfile::tempdir().unwrap();
let parent = tmp.path().join("private");
fs::create_dir(&parent).unwrap();
fs::set_permissions(&parent, fs::Permissions::from_mode(0o600)).unwrap();
// Root can search this directory despite its mode, so run the denial
// assertion only when the test process is subject to these permissions.
let can_search = run_upload_shell("test -x private", tmp.path(), None)
.status
.success();
let output = run_upload_shell(
&build_upload_destination_probe_cmd("private/nested/config.json"),
tmp.path(),
None,
);
fs::set_permissions(&parent, fs::Permissions::from_mode(0o700)).unwrap();
if !can_search {
assert!(!output.status.success());
assert!(output.stdout.is_empty());
assert!(String::from_utf8_lossy(&output.stderr).contains("cannot inspect"));
}
}
#[cfg(unix)]
#[test]
fn upload_rejects_detected_destination_type_changes_before_extraction() {
for initial_kind in [
UploadDestinationKind::Missing,
UploadDestinationKind::File,
UploadDestinationKind::Directory,
] {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("config.json");
fs::write(&source, b"configuration").unwrap();
let destination = tmp.path().join("destination");
match initial_kind {
UploadDestinationKind::Directory => fs::create_dir(&destination).unwrap(),
UploadDestinationKind::File => fs::write(&destination, b"old").unwrap(),
UploadDestinationKind::Missing => {}
}
let (command, archive) =
prepare_file_upload_command(&source, "destination", tmp.path());
if initial_kind == UploadDestinationKind::Directory {
fs::remove_dir(&destination).unwrap();
fs::write(&destination, b"replacement").unwrap();
} else {
if initial_kind == UploadDestinationKind::File {
fs::remove_file(&destination).unwrap();
}
fs::create_dir(&destination).unwrap();
fs::write(destination.join("keep.txt"), b"keep").unwrap();
}
let output = run_upload_shell(&command, tmp.path(), Some(&archive));
assert_eq!(
output.status.code(),
Some(UPLOAD_DESTINATION_CHANGED_EXIT_CODE)
);
assert!(String::from_utf8_lossy(&output.stderr).contains("changed type"));
if initial_kind == UploadDestinationKind::Directory {
assert_eq!(fs::read(&destination).unwrap(), b"replacement");
} else {
assert_eq!(fs::read(destination.join("keep.txt")).unwrap(), b"keep");
assert!(!destination.join("config.json").exists());
}
}
assert!(!sync_error_is_retryable(&miette::miette!(
"upload destination changed type before extraction; retry the upload"
)));
}
#[cfg(unix)]
#[test]
fn upload_file_list_keeps_selection_and_directory_prefix() {
let tmp = tempfile::tempdir().unwrap();
let source = tmp.path().join("source");
fs::create_dir(&source).unwrap();
fs::write(source.join("selected.json"), b"selected").unwrap();
fs::write(source.join("ignored.json"), b"ignored").unwrap();
let mut archive = Vec::new();
write_upload_archive(
&mut archive,
UploadSource::FileList {
base_dir: source,
files: vec!["selected.json".into()],
archive_prefix: Some("source".into()),
},
)
.unwrap();
let output = run_upload_shell(
&build_upload_extract_cmd("destination", None),
tmp.path(),
Some(&archive),
);
assert!(
output.status.success(),
"{}",
String::from_utf8_lossy(&output.stderr)
);
assert_eq!(
fs::read(tmp.path().join("destination/source/selected.json")).unwrap(),
b"selected"
);
assert!(!tmp.path().join("destination/source/ignored.json").exists());
}
#[test]
fn upload_destination_probe_preserves_explicit_directory_and_top_level_rules() {
for path in ["/sandbox", "/sandbox/", "relative/", "/"] {
assert!(!file_upload_destination_needs_probe(path), "{path}");
}
for path in ["/sandbox/config.json", "relative", ".", "nested/directory"] {
assert!(file_upload_destination_needs_probe(path), "{path}");
}
}
#[test]
fn directory_upload_prefix_uses_basename_for_named_directories() {
assert_eq!(
directory_upload_prefix(Path::new("/tmp/upload-test")),
std::ffi::OsString::from("upload-test")
);
assert_eq!(
directory_upload_prefix(Path::new("foo")),
std::ffi::OsString::from("foo")
);
assert_eq!(
directory_upload_prefix(Path::new("./parent/nested")),
std::ffi::OsString::from("nested")
);
}
#[test]
fn directory_upload_prefix_falls_back_to_dot_for_basename_less_paths() {
assert_eq!(
directory_upload_prefix(Path::new(".")),
std::ffi::OsString::from(".")
);
assert_eq!(
directory_upload_prefix(Path::new("/")),
std::ffi::OsString::from(".")
);
}
#[test]
fn file_list_archive_prefix_uses_named_directory_basename() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let source = tmpdir.path().join("source-dir");
let file = tmpdir.path().join("file.txt");
fs::create_dir_all(&source).expect("create source dir");
fs::write(&file, "file").expect("write file");
assert_eq!(
file_list_archive_prefix(&source),
Some(PathBuf::from("source-dir"))
);
assert_eq!(file_list_archive_prefix(Path::new(".")), None);
assert_eq!(file_list_archive_prefix(&file), None);
}
#[derive(Debug)]
struct UploadArchiveEntry {
path: String,
#[cfg_attr(not(unix), allow(dead_code))]
entry_type: tar::EntryType,
#[cfg_attr(not(unix), allow(dead_code))]
link_name: Option<String>,
}
fn upload_archive_entries(source: UploadSource) -> Vec<UploadArchiveEntry> {
let mut bytes = Vec::new();
write_upload_archive(&mut bytes, source).expect("write upload archive");
let mut archive = tar::Archive::new(std::io::Cursor::new(bytes));
let entries = archive.entries().expect("read archive entries");
let mut entries = entries
.map(|entry| {
let entry = entry.expect("read archive entry");
let path = entry
.path()
.expect("read archive path")
.to_string_lossy()
.into_owned();
let entry_type = entry.header().entry_type();
let link_name = entry
.link_name()
.expect("read archive link")
.map(|link| link.to_string_lossy().into_owned());
UploadArchiveEntry {
path,
entry_type,
link_name,
}
})
.collect::<Vec<_>>();
entries.sort_by(|left, right| left.path.cmp(&right.path));
entries
}
fn upload_archive_paths(source: UploadSource) -> Vec<String> {
let mut paths = upload_archive_entries(source)
.into_iter()
.map(|entry| entry.path)
.collect::<Vec<_>>();
paths.sort();
paths
}
#[test]
fn file_list_archive_preserves_directory_prefix_when_requested() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let base_dir = tmpdir.path().join("nested");
fs::create_dir_all(base_dir.join("inner")).expect("create dirs");
fs::write(base_dir.join("file.txt"), "file").expect("write file");
fs::write(base_dir.join("inner/child.txt"), "child").expect("write child");
let paths = upload_archive_paths(UploadSource::FileList {
base_dir,
files: vec!["file.txt".into(), "inner/child.txt".into()],
archive_prefix: Some(PathBuf::from("nested")),
});
assert_eq!(paths, vec!["nested/file.txt", "nested/inner/child.txt"]);
}
#[test]
fn file_list_archive_stays_flat_without_directory_prefix() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let base_dir = tmpdir.path().join("nested");
fs::create_dir_all(base_dir.join("inner")).expect("create dirs");
fs::write(base_dir.join("file.txt"), "file").expect("write file");
fs::write(base_dir.join("inner/child.txt"), "child").expect("write child");
let paths = upload_archive_paths(UploadSource::FileList {
base_dir,
files: vec!["file.txt".into(), "inner/child.txt".into()],
archive_prefix: None,
});
assert_eq!(paths, vec!["file.txt", "inner/child.txt"]);
}
#[test]
fn single_directory_archive_preserves_directory_basename() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let source = tmpdir.path().join("source-dir");
fs::create_dir_all(source.join("inner")).expect("create dirs");
fs::write(source.join("file.txt"), "file").expect("write file");
fs::write(source.join("inner/child.txt"), "child").expect("write child");
let paths = upload_archive_paths(UploadSource::SinglePath {
local_path: source,
tar_name: "source-dir".into(),
});
assert!(paths.contains(&"source-dir/file.txt".to_string()));
assert!(paths.contains(&"source-dir/inner/child.txt".to_string()));
assert!(paths.iter().all(|path| path.starts_with("source-dir/")));
}
#[cfg(unix)]
#[test]
fn single_directory_archive_preserves_symlink_entries() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let source = tmpdir.path().join("source-dir");
fs::create_dir_all(source.join("real-dir")).expect("create dirs");
fs::write(source.join("real-dir/file.txt"), "file").expect("write file");
std::os::unix::fs::symlink("real-dir", source.join("link-dir")).expect("create symlink");
let entries = upload_archive_entries(UploadSource::SinglePath {
local_path: source,
tar_name: "source-dir".into(),
});
let symlink = entries
.iter()
.find(|entry| entry.path == "source-dir/link-dir")
.expect("symlink archive entry");
assert_eq!(symlink.entry_type, tar::EntryType::Symlink);
assert_eq!(symlink.link_name.as_deref(), Some("real-dir"));
assert!(
entries
.iter()
.all(|entry| entry.path != "source-dir/link-dir/file.txt"),
"symlink target should not be expanded into the archive: {entries:?}"
);
}
#[cfg(unix)]
#[test]
fn file_list_archive_preserves_symlink_entries() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let base_dir = tmpdir.path().join("nested");
fs::create_dir_all(base_dir.join("real-dir")).expect("create dirs");
fs::write(base_dir.join("real-dir/file.txt"), "file").expect("write file");
std::os::unix::fs::symlink("real-dir", base_dir.join("link-dir")).expect("create symlink");
let entries = upload_archive_entries(UploadSource::FileList {
base_dir,
files: vec!["link-dir".into()],
archive_prefix: Some(PathBuf::from("nested")),
});
assert_eq!(entries.len(), 1, "unexpected archive entries: {entries:?}");
let symlink = &entries[0];
assert_eq!(symlink.path, "nested/link-dir");
assert_eq!(symlink.entry_type, tar::EntryType::Symlink);
assert_eq!(symlink.link_name.as_deref(), Some("real-dir"));
}
#[cfg(unix)]
#[test]
fn single_symlink_archive_preserves_link_target() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
fs::write(tmpdir.path().join("target.txt"), "target").expect("write target");
let link = tmpdir.path().join("link.txt");
std::os::unix::fs::symlink("target.txt", &link).expect("create symlink");
let entries = upload_archive_entries(UploadSource::SinglePath {
local_path: link,
tar_name: "uploaded-link.txt".into(),
});
assert_eq!(entries.len(), 1, "unexpected archive entries: {entries:?}");
assert_eq!(entries[0].path, "uploaded-link.txt");
assert_eq!(entries[0].entry_type, tar::EntryType::Symlink);
assert_eq!(entries[0].link_name.as_deref(), Some("target.txt"));
}
#[cfg(unix)]
#[test]
fn dangling_single_symlink_archive_preserves_link_target() {
let tmpdir = tempfile::tempdir().expect("create tmpdir");
let link = tmpdir.path().join("dangling-link.txt");
std::os::unix::fs::symlink("missing.txt", &link).expect("create symlink");
let entries = upload_archive_entries(UploadSource::SinglePath {
local_path: link,
tar_name: "dangling-link.txt".into(),
});
assert_eq!(entries.len(), 1, "unexpected archive entries: {entries:?}");
assert_eq!(entries[0].path, "dangling-link.txt");
assert_eq!(entries[0].entry_type, tar::EntryType::Symlink);
assert_eq!(entries[0].link_name.as_deref(), Some("missing.txt"));
}
#[test]
fn split_sandbox_path_handles_root_and_bare_names() {
// File directly under root
assert_eq!(split_sandbox_path("/.bashrc"), ("/", ".bashrc"));
// No directory component at all
assert_eq!(split_sandbox_path("file.txt"), (".", "file.txt"));
}
}