#![forbid(unsafe_code)]
use super::selection::{resolve_host_jobs, HostSelection};
use super::{
apply_overrides, build_connection_config, load, resolve_config_path, validate_command_length,
};
use crate::cli::OutputFormat;
use crate::errors::{SshCliError, SshCliResult};
use crate::output;
use crate::ssh::client::{ExecutionOutput, SshClient, SshClientTrait};
use crate::ssh::packing::{append_description, pack_su, pack_sudo};
use crate::vps::model::{effective_limit, VpsRecord};
use anyhow::Result;
use secrecy::SecretString;
use std::path::PathBuf;
#[derive(Debug, Default, Clone)]
pub struct ExecOptions {
pub password: Option<SecretString>,
pub sudo_password: Option<SecretString>,
pub su_password: Option<SecretString>,
pub timeout: Option<crate::domain::TimeoutMs>,
pub key: Option<String>,
pub key_passphrase: Option<SecretString>,
pub use_agent: bool,
pub agent_socket: Option<String>,
pub description: Option<String>,
pub replace_host_key: bool,
pub disable_sudo: bool,
pub steps: Vec<crate::domain::RemoteCommand>,
}
#[derive(Clone, Copy)]
enum ExecKind {
Plain,
Sudo,
Su,
}
#[derive(Debug, Clone)]
pub struct HostExecResult {
pub name: String,
pub ok: bool,
pub exit_code: Option<i32>,
pub stdout: String,
pub stderr: String,
pub duration_ms: u64,
pub error: Option<String>,
}
async fn finish_execution_output(
client: Box<dyn SshClientTrait>,
result: SshCliResult<ExecutionOutput>,
format: OutputFormat,
json: bool,
) -> Result<()> {
let _ = client.disconnect().await;
let output = result?;
if format == OutputFormat::Json || json {
output::print_execution_output_json(&output)?;
} else {
output::print_execution_output(&output);
}
if let Some(code) = output.exit_code {
if code != 0 {
return Err(SshCliError::CommandFailed {
exit_code: code,
stderr: output.stderr,
}
.into());
}
}
Ok(())
}
fn cancelled_err() -> anyhow::Error {
anyhow::anyhow!(crate::i18n::t(crate::i18n::Message::OperationCancelled))
}
fn expect_single(selection: HostSelection) -> Result<String> {
match selection {
HostSelection::Single(name) => Ok(name.into_inner()),
_ => Err(SshCliError::InvalidArgument(
"internal: expected single-host selection for non-batch exec".into(),
)
.into()),
}
}
#[allow(clippy::too_many_arguments)]
pub async fn run_exec(
selection: HostSelection,
command: &str,
config_override: Option<PathBuf>,
format: OutputFormat,
json: bool,
opts: ExecOptions,
) -> Result<()> {
if crate::signals::should_stop() {
return Err(cancelled_err());
}
if selection.is_batch() {
return run_exec_all(
&selection,
command,
config_override,
format,
json,
opts,
ExecKind::Plain,
)
.await;
}
let vps_name = expect_single(selection)?;
let path = resolve_config_path(config_override.as_deref())?;
let mut file = load(&path)?;
let mut vps = file
.hosts
.remove(&vps_name)
.ok_or(SshCliError::VpsNotFound(vps_name))?;
apply_overrides(
&mut vps,
opts.password,
opts.sudo_password,
opts.su_password,
opts.timeout,
opts.key,
opts.key_passphrase,
opts.use_agent,
opts.agent_socket,
);
let cmd = append_description(command, opts.description.as_deref());
validate_command_length(&cmd, vps.max_command_chars.wire())?;
for s in &opts.steps {
validate_command_length(s.as_str(), vps.max_command_chars.wire())?;
}
let cfg = build_connection_config(&vps, Some(&path), opts.replace_host_key);
let client: Box<dyn SshClientTrait> = <SshClient as SshClientTrait>::connect(cfg).await?;
run_exec_with_client_steps(&vps, &cmd, &opts.steps, client, format, json).await
}
pub async fn run_exec_with_client(
vps: &VpsRecord,
command: &str,
client: Box<dyn SshClientTrait>,
format: OutputFormat,
json: bool,
) -> Result<()> {
run_exec_with_client_steps(vps, command, &[], client, format, json).await
}
pub async fn run_exec_with_client_steps(
vps: &VpsRecord,
command: &str,
steps: &[crate::domain::RemoteCommand],
mut client: Box<dyn SshClientTrait>,
format: OutputFormat,
json: bool,
) -> Result<()> {
if crate::signals::should_stop() {
return Err(cancelled_err());
}
let max_out = effective_limit(vps.max_output_chars.wire());
let mut cmds: Vec<&str> = Vec::with_capacity(1 + steps.len());
cmds.push(command);
for s in steps {
cmds.push(s.as_str());
}
let mut last_output = None;
let mut failed: Option<(i32, String)> = None;
for (i, cmd) in cmds.iter().enumerate() {
if crate::signals::should_stop() {
let _ = client.disconnect().await;
return Err(cancelled_err());
}
tracing::debug!(step = i, "exec multi-cmd step");
match client.run_command(cmd, max_out, None).await {
Ok(output) => {
if format == OutputFormat::Json || json {
let mut v = serde_json::to_value(crate::json_wire::ExecutionJson::from(&output))
.unwrap_or_else(|_| serde_json::json!({}));
if let Some(obj) = v.as_object_mut() {
obj.insert("step".into(), serde_json::json!(i));
obj.insert("command".into(), serde_json::json!(cmd));
}
crate::output::print_json_value(&v)?;
} else if cmds.len() > 1 {
crate::output::write_line_fmt(format_args!("--- step {i}: {cmd} ---"))?;
crate::output::print_execution_output(&output);
} else {
last_output = Some(output.clone());
}
if let Some(code) = output.exit_code {
if code != 0 && failed.is_none() {
failed = Some((code, output.stderr.clone()));
}
}
if cmds.len() == 1 {
last_output = Some(output);
}
}
Err(e) => {
let _ = client.disconnect().await;
return Err(e.into());
}
}
}
let _ = client.disconnect().await;
if let Some((code, stderr)) = failed {
return Err(SshCliError::CommandFailed {
exit_code: code,
stderr,
}
.into());
}
if let Some(output) = last_output {
if format == OutputFormat::Json || json {
if cmds.len() == 1 {
crate::output::print_execution_output_json(&output)?;
}
} else if cmds.len() == 1 {
crate::output::print_execution_output(&output);
}
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn run_sudo_exec(
selection: HostSelection,
command: &str,
config_override: Option<PathBuf>,
format: OutputFormat,
json: bool,
opts: ExecOptions,
) -> Result<()> {
if crate::signals::should_stop() {
return Err(cancelled_err());
}
if selection.is_batch() {
return run_exec_all(
&selection,
command,
config_override,
format,
json,
opts,
ExecKind::Sudo,
)
.await;
}
let vps_name = expect_single(selection)?;
let path = resolve_config_path(config_override.as_deref())?;
let mut file = load(&path)?;
let mut vps = file
.hosts
.remove(&vps_name)
.ok_or(SshCliError::VpsNotFound(vps_name))?;
apply_overrides(
&mut vps,
opts.password,
opts.sudo_password,
opts.su_password,
opts.timeout,
opts.key,
opts.key_passphrase,
opts.use_agent,
opts.agent_socket,
);
if opts.disable_sudo || vps.disable_sudo {
return Err(SshCliError::SudoDisabled.into());
}
let cmd = append_description(command, opts.description.as_deref());
validate_command_length(&cmd, vps.max_command_chars.wire())?;
let cfg = build_connection_config(&vps, Some(&path), opts.replace_host_key);
let client: Box<dyn SshClientTrait> = <SshClient as SshClientTrait>::connect(cfg).await?;
run_sudo_exec_with_client(&vps, &cmd, client, format, json).await
}
pub async fn run_sudo_exec_with_client(
vps: &VpsRecord,
command: &str,
client: Box<dyn SshClientTrait>,
format: OutputFormat,
json: bool,
) -> Result<()> {
if crate::signals::should_stop() {
return Err(cancelled_err());
}
if vps.disable_sudo {
return Err(SshCliError::SudoDisabled.into());
}
let mut pack = pack_sudo(command, vps.sudo_password.as_ref());
let max_out = effective_limit(vps.max_output_chars.wire());
let stdin = pack.take_stdin();
let mut client = client;
let result = client.run_command(&pack.command, max_out, stdin).await;
finish_execution_output(client, result, format, json).await
}
#[allow(clippy::too_many_arguments)]
pub async fn run_su_exec(
selection: HostSelection,
command: &str,
config_override: Option<PathBuf>,
format: OutputFormat,
json: bool,
opts: ExecOptions,
) -> Result<()> {
if crate::signals::should_stop() {
return Err(cancelled_err());
}
if selection.is_batch() {
return run_exec_all(
&selection,
command,
config_override,
format,
json,
opts,
ExecKind::Su,
)
.await;
}
let vps_name = expect_single(selection)?;
let path = resolve_config_path(config_override.as_deref())?;
let mut file = load(&path)?;
let mut vps = file
.hosts
.remove(&vps_name)
.ok_or(SshCliError::VpsNotFound(vps_name))?;
apply_overrides(
&mut vps,
opts.password,
opts.sudo_password,
opts.su_password,
opts.timeout,
opts.key,
opts.key_passphrase,
opts.use_agent,
opts.agent_socket,
);
if opts.disable_sudo || vps.disable_sudo {
return Err(SshCliError::SudoDisabled.into());
}
let su_password = vps.su_password.take().ok_or(SshCliError::SuPasswordMissing)?;
let cmd = append_description(command, opts.description.as_deref());
validate_command_length(&cmd, vps.max_command_chars.wire())?;
let mut pack = pack_su(&cmd, &su_password);
let cfg = build_connection_config(&vps, Some(&path), opts.replace_host_key);
let mut client: Box<dyn SshClientTrait> =
<SshClient as SshClientTrait>::connect(cfg).await?;
let max_out = effective_limit(vps.max_output_chars.wire());
let stdin = pack.take_stdin();
let result = client.run_command(&pack.command, max_out, stdin).await;
finish_execution_output(client, result, format, json).await
}
#[allow(clippy::too_many_arguments)]
async fn run_exec_all(
selection: &HostSelection,
command: &str,
config_override: Option<PathBuf>,
format: OutputFormat,
json: bool,
opts: ExecOptions,
kind: ExecKind,
) -> Result<()> {
let path = resolve_config_path(config_override.as_deref())?;
let file = load(&path)?;
let jobs = resolve_host_jobs(selection, &file)?;
let limit = crate::concurrency::effective_limit();
let cmd_base = command.to_string();
let path_c = path.clone();
let opts_c = std::sync::Arc::new(opts);
let replace = opts_c.replace_host_key;
let total_jobs = jobs.len();
tracing::info!(
hosts = jobs.len(),
max_concurrency = limit,
fail_fast = crate::concurrency::fail_fast_enabled(),
kind = ?match kind {
ExecKind::Plain => "exec",
ExecKind::Sudo => "sudo-exec",
ExecKind::Su => "su-exec",
},
"multi-host exec fan-out"
);
let results = crate::concurrency::map_bounded_with(
jobs,
limit,
move |(name, mut vps)| {
let cmd_base = cmd_base.clone();
let path_c = path_c.clone();
let opts_arc = std::sync::Arc::clone(&opts_c);
async move {
let mut opts = (*opts_arc).clone();
if crate::signals::should_stop() {
return HostExecResult {
name,
ok: false,
exit_code: None,
stdout: String::new(),
stderr: "cancelled".into(),
duration_ms: 0,
error: Some("operation cancelled by signal".into()),
};
}
apply_overrides(
&mut vps,
opts.password.take(),
opts.sudo_password.take(),
opts.su_password.take(),
opts.timeout,
opts.key.take(),
opts.key_passphrase.take(),
opts.use_agent,
opts.agent_socket.take(),
);
let cmd = append_description(&cmd_base, opts.description.as_deref());
if let Err(e) = validate_command_length(&cmd, vps.max_command_chars.wire()) {
return HostExecResult {
name,
ok: false,
exit_code: None,
stdout: String::new(),
stderr: e.to_string(),
duration_ms: 0,
error: Some(e.to_string()),
};
}
match kind {
ExecKind::Sudo | ExecKind::Su if opts.disable_sudo || vps.disable_sudo => {
return HostExecResult {
name,
ok: false,
exit_code: None,
stdout: String::new(),
stderr: "sudo/su disabled".into(),
duration_ms: 0,
error: Some("sudo/su disabled".into()),
};
}
_ => {}
}
let start = std::time::Instant::now();
let run = async {
let cfg = build_connection_config(&vps, Some(&path_c), replace);
let mut client: Box<dyn SshClientTrait> =
<SshClient as SshClientTrait>::connect(cfg).await?;
let max_out = effective_limit(vps.max_output_chars.wire());
let output = match kind {
ExecKind::Plain => client.run_command(&cmd, max_out, None).await?,
ExecKind::Sudo => {
let mut pack = pack_sudo(&cmd, vps.sudo_password.as_ref());
let stdin = pack.take_stdin();
client.run_command(&pack.command, max_out, stdin).await?
}
ExecKind::Su => {
let su_pw = vps
.su_password
.take()
.ok_or(SshCliError::SuPasswordMissing)?;
let mut pack = pack_su(&cmd, &su_pw);
let stdin = pack.take_stdin();
client.run_command(&pack.command, max_out, stdin).await?
}
};
let _ = client.disconnect().await;
Ok::<_, SshCliError>(output)
}
.await;
let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
match run {
Ok(output) => {
let code_ok = output.exit_code.unwrap_or(0) == 0;
HostExecResult {
name,
ok: code_ok,
exit_code: output.exit_code,
stdout: output.stdout,
stderr: output.stderr,
duration_ms,
error: if code_ok {
None
} else {
Some(format!(
"exit {}",
output.exit_code.unwrap_or(-1)
))
},
}
}
Err(e) => HostExecResult {
name,
ok: false,
exit_code: None,
stdout: String::new(),
stderr: e.to_string(),
duration_ms,
error: Some(e.to_string()),
},
}
}
},
|h: &HostExecResult| !h.ok,
)
.await;
let mut host_results = Vec::with_capacity(total_jobs.max(results.len()));
let mut failures = 0usize;
let mut seen = std::collections::BTreeSet::new();
for r in results {
match r.outcome {
Ok(h) => {
if !h.ok {
failures += 1;
}
seen.insert(r.index);
host_results.push(h);
}
Err(e) if e.is_panic() => std::panic::resume_unwind(e.into_panic()),
Err(e) => {
failures += 1;
seen.insert(r.index);
host_results.push(HostExecResult {
name: format!("task-{}", r.index),
ok: false,
exit_code: None,
stdout: String::new(),
stderr: e.to_string(),
duration_ms: 0,
error: Some(e.to_string()),
});
}
}
}
if crate::concurrency::fail_fast_enabled() && host_results.len() < total_jobs {
let skipped = total_jobs - host_results.len();
for i in 0..total_jobs {
if !seen.contains(&i) {
host_results.push(HostExecResult {
name: format!("skipped-{i}"),
ok: false,
exit_code: None,
stdout: String::new(),
stderr: "skipped (fail-fast)".into(),
duration_ms: 0,
error: Some("skipped (fail-fast)".into()),
});
}
}
let _ = skipped;
}
let as_json = format == OutputFormat::Json || json;
output::print_exec_batch(&host_results, limit, as_json)?;
if failures > 0 || host_results.iter().any(|h| !h.ok) {
let failed = host_results.iter().filter(|h| !h.ok).count();
return Err(SshCliError::Config(format!(
"{failed}/{} hosts failed multi-host exec",
host_results.len()
))
.into());
}
Ok(())
}