ssh_session.rs

  1use crate::{
  2    json_log::LogRecord,
  3    protocol::{
  4        message_len_from_buffer, read_message_with_len, write_message, MessageId, MESSAGE_LEN_SIZE,
  5    },
  6};
  7use anyhow::{anyhow, Context as _, Result};
  8use collections::HashMap;
  9use futures::{
 10    channel::{mpsc, oneshot},
 11    future::BoxFuture,
 12    select_biased, AsyncReadExt as _, AsyncWriteExt as _, Future, FutureExt as _, StreamExt as _,
 13};
 14use gpui::{AppContext, AsyncAppContext, Model, SemanticVersion};
 15use parking_lot::Mutex;
 16use rpc::{
 17    proto::{self, build_typed_envelope, Envelope, EnvelopedMessage, PeerId, RequestMessage},
 18    EntityMessageSubscriber, ProtoClient, ProtoMessageHandlerSet, RpcError,
 19};
 20use smol::{
 21    fs,
 22    process::{self, Stdio},
 23};
 24use std::{
 25    any::TypeId,
 26    ffi::OsStr,
 27    path::{Path, PathBuf},
 28    sync::{
 29        atomic::{AtomicU32, Ordering::SeqCst},
 30        Arc,
 31    },
 32    time::Instant,
 33};
 34use tempfile::TempDir;
 35
 36#[derive(
 37    Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy, serde::Serialize, serde::Deserialize,
 38)]
 39pub struct SshProjectId(pub u64);
 40
 41#[derive(Clone)]
 42pub struct SshSocket {
 43    connection_options: SshConnectionOptions,
 44    socket_path: PathBuf,
 45}
 46
 47pub struct SshSession {
 48    next_message_id: AtomicU32,
 49    response_channels: ResponseChannels, // Lock
 50    outgoing_tx: mpsc::UnboundedSender<Envelope>,
 51    spawn_process_tx: mpsc::UnboundedSender<SpawnRequest>,
 52    client_socket: Option<SshSocket>,
 53    state: Mutex<ProtoMessageHandlerSet>, // Lock
 54}
 55
 56struct SshClientState {
 57    socket: SshSocket,
 58    _master_process: process::Child,
 59    _temp_dir: TempDir,
 60}
 61
 62#[derive(Debug, Clone, PartialEq, Eq)]
 63pub struct SshConnectionOptions {
 64    pub host: String,
 65    pub username: Option<String>,
 66    pub port: Option<u16>,
 67    pub password: Option<String>,
 68}
 69
 70impl SshConnectionOptions {
 71    pub fn ssh_url(&self) -> String {
 72        let mut result = String::from("ssh://");
 73        if let Some(username) = &self.username {
 74            result.push_str(username);
 75            result.push('@');
 76        }
 77        result.push_str(&self.host);
 78        if let Some(port) = self.port {
 79            result.push(':');
 80            result.push_str(&port.to_string());
 81        }
 82        result
 83    }
 84
 85    fn scp_url(&self) -> String {
 86        if let Some(username) = &self.username {
 87            format!("{}@{}", username, self.host)
 88        } else {
 89            self.host.clone()
 90        }
 91    }
 92
 93    pub fn connection_string(&self) -> String {
 94        let host = if let Some(username) = &self.username {
 95            format!("{}@{}", username, self.host)
 96        } else {
 97            self.host.clone()
 98        };
 99        if let Some(port) = &self.port {
100            format!("{}:{}", host, port)
101        } else {
102            host
103        }
104    }
105}
106
107struct SpawnRequest {
108    command: String,
109    process_tx: oneshot::Sender<process::Child>,
110}
111
112#[derive(Copy, Clone, Debug)]
113pub struct SshPlatform {
114    pub os: &'static str,
115    pub arch: &'static str,
116}
117
118pub trait SshClientDelegate {
119    fn ask_password(
120        &self,
121        prompt: String,
122        cx: &mut AsyncAppContext,
123    ) -> oneshot::Receiver<Result<String>>;
124    fn remote_server_binary_path(&self, cx: &mut AsyncAppContext) -> Result<PathBuf>;
125    fn get_server_binary(
126        &self,
127        platform: SshPlatform,
128        cx: &mut AsyncAppContext,
129    ) -> oneshot::Receiver<Result<(PathBuf, SemanticVersion)>>;
130    fn set_status(&self, status: Option<&str>, cx: &mut AsyncAppContext);
131}
132
133type ResponseChannels = Mutex<HashMap<MessageId, oneshot::Sender<(Envelope, oneshot::Sender<()>)>>>;
134
135impl SshSession {
136    pub async fn client(
137        connection_options: SshConnectionOptions,
138        delegate: Arc<dyn SshClientDelegate>,
139        cx: &mut AsyncAppContext,
140    ) -> Result<Arc<Self>> {
141        let client_state = SshClientState::new(connection_options, delegate.clone(), cx).await?;
142
143        let platform = client_state.query_platform().await?;
144        let (local_binary_path, version) = delegate.get_server_binary(platform, cx).await??;
145        let remote_binary_path = delegate.remote_server_binary_path(cx)?;
146        client_state
147            .ensure_server_binary(
148                &delegate,
149                &local_binary_path,
150                &remote_binary_path,
151                version,
152                cx,
153            )
154            .await?;
155
156        let (spawn_process_tx, mut spawn_process_rx) = mpsc::unbounded::<SpawnRequest>();
157        let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded::<Envelope>();
158        let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();
159
160        let socket = client_state.socket.clone();
161        run_cmd(socket.ssh_command(&remote_binary_path).arg("version")).await?;
162
163        let mut remote_server_child = socket
164            .ssh_command(format!(
165                "RUST_LOG={} RUST_BACKTRACE={} {:?} run",
166                std::env::var("RUST_LOG").unwrap_or_default(),
167                std::env::var("RUST_BACKTRACE").unwrap_or_default(),
168                remote_binary_path,
169            ))
170            .spawn()
171            .context("failed to spawn remote server")?;
172        let mut child_stderr = remote_server_child.stderr.take().unwrap();
173        let mut child_stdout = remote_server_child.stdout.take().unwrap();
174        let mut child_stdin = remote_server_child.stdin.take().unwrap();
175
176        let executor = cx.background_executor().clone();
177        executor.clone().spawn(async move {
178            let mut stdin_buffer = Vec::new();
179            let mut stdout_buffer = Vec::new();
180            let mut stderr_buffer = Vec::new();
181            let mut stderr_offset = 0;
182
183            loop {
184                stdout_buffer.resize(MESSAGE_LEN_SIZE, 0);
185                stderr_buffer.resize(stderr_offset + 1024, 0);
186
187                select_biased! {
188                    outgoing = outgoing_rx.next().fuse() => {
189                        let Some(outgoing) = outgoing else {
190                            return anyhow::Ok(());
191                        };
192
193                        write_message(&mut child_stdin, &mut stdin_buffer, outgoing).await?;
194                    }
195
196                    request = spawn_process_rx.next().fuse() => {
197                        let Some(request) = request else {
198                            return Ok(());
199                        };
200
201                        log::info!("spawn process: {:?}", request.command);
202                        let child = client_state.socket
203                            .ssh_command(&request.command)
204                            .spawn()
205                            .context("failed to create channel")?;
206                        request.process_tx.send(child).ok();
207                    }
208
209                    result = child_stdout.read(&mut stdout_buffer).fuse() => {
210                        match result {
211                            Ok(len) => {
212                                if len == 0 {
213                                    child_stdin.close().await?;
214                                    let status = remote_server_child.status().await?;
215                                    if !status.success() {
216                                        log::info!("channel exited with status: {status:?}");
217                                    }
218                                    return Ok(());
219                                }
220
221                                if len < stdout_buffer.len() {
222                                    child_stdout.read_exact(&mut stdout_buffer[len..]).await?;
223                                }
224
225                                let message_len = message_len_from_buffer(&stdout_buffer);
226                                match read_message_with_len(&mut child_stdout, &mut stdout_buffer, message_len).await {
227                                    Ok(envelope) => {
228                                        incoming_tx.unbounded_send(envelope).ok();
229                                    }
230                                    Err(error) => {
231                                        log::error!("error decoding message {error:?}");
232                                    }
233                                }
234                            }
235                            Err(error) => {
236                                Err(anyhow!("error reading stdout: {error:?}"))?;
237                            }
238                        }
239                    }
240
241                    result = child_stderr.read(&mut stderr_buffer[stderr_offset..]).fuse() => {
242                        match result {
243                            Ok(len) => {
244                                stderr_offset += len;
245                                let mut start_ix = 0;
246                                while let Some(ix) = stderr_buffer[start_ix..stderr_offset].iter().position(|b| b == &b'\n') {
247                                    let line_ix = start_ix + ix;
248                                    let content = &stderr_buffer[start_ix..line_ix];
249                                    start_ix = line_ix + 1;
250                                    if let Ok(mut record) = serde_json::from_slice::<LogRecord>(content) {
251                                        record.message = format!("(remote) {}", record.message);
252                                        record.log(log::logger())
253                                    } else {
254                                        eprintln!("(remote) {}", String::from_utf8_lossy(content));
255                                    }
256                                }
257                                stderr_buffer.drain(0..start_ix);
258                                stderr_offset -= start_ix;
259                            }
260                            Err(error) => {
261                                Err(anyhow!("error reading stderr: {error:?}"))?;
262                            }
263                        }
264                    }
265                }
266            }
267        }).detach();
268
269        cx.update(|cx| Self::new(incoming_rx, outgoing_tx, spawn_process_tx, Some(socket), cx))
270    }
271
272    pub fn server(
273        incoming_rx: mpsc::UnboundedReceiver<Envelope>,
274        outgoing_tx: mpsc::UnboundedSender<Envelope>,
275        cx: &AppContext,
276    ) -> Arc<SshSession> {
277        let (tx, _rx) = mpsc::unbounded();
278        Self::new(incoming_rx, outgoing_tx, tx, None, cx)
279    }
280
281    #[cfg(any(test, feature = "test-support"))]
282    pub fn fake(
283        client_cx: &mut gpui::TestAppContext,
284        server_cx: &mut gpui::TestAppContext,
285    ) -> (Arc<Self>, Arc<Self>) {
286        let (server_to_client_tx, server_to_client_rx) = mpsc::unbounded();
287        let (client_to_server_tx, client_to_server_rx) = mpsc::unbounded();
288        let (tx, _rx) = mpsc::unbounded();
289        (
290            client_cx.update(|cx| {
291                Self::new(
292                    server_to_client_rx,
293                    client_to_server_tx,
294                    tx.clone(),
295                    None, // todo()
296                    cx,
297                )
298            }),
299            server_cx.update(|cx| {
300                Self::new(
301                    client_to_server_rx,
302                    server_to_client_tx,
303                    tx.clone(),
304                    None,
305                    cx,
306                )
307            }),
308        )
309    }
310
311    fn new(
312        mut incoming_rx: mpsc::UnboundedReceiver<Envelope>,
313        outgoing_tx: mpsc::UnboundedSender<Envelope>,
314        spawn_process_tx: mpsc::UnboundedSender<SpawnRequest>,
315        client_socket: Option<SshSocket>,
316        cx: &AppContext,
317    ) -> Arc<SshSession> {
318        let this = Arc::new(Self {
319            next_message_id: AtomicU32::new(0),
320            response_channels: ResponseChannels::default(),
321            outgoing_tx,
322            spawn_process_tx,
323            client_socket,
324            state: Default::default(),
325        });
326
327        cx.spawn(|cx| {
328            let this = this.clone();
329            async move {
330                let peer_id = PeerId { owner_id: 0, id: 0 };
331                while let Some(incoming) = incoming_rx.next().await {
332                    if let Some(request_id) = incoming.responding_to {
333                        let request_id = MessageId(request_id);
334                        let sender = this.response_channels.lock().remove(&request_id);
335                        if let Some(sender) = sender {
336                            let (tx, rx) = oneshot::channel();
337                            if incoming.payload.is_some() {
338                                sender.send((incoming, tx)).ok();
339                            }
340                            rx.await.ok();
341                        }
342                    } else if let Some(envelope) =
343                        build_typed_envelope(peer_id, Instant::now(), incoming)
344                    {
345                        let type_name = envelope.payload_type_name();
346                        if let Some(future) = ProtoMessageHandlerSet::handle_message(
347                            &this.state,
348                            envelope,
349                            this.clone().into(),
350                            cx.clone(),
351                        ) {
352                            log::debug!("ssh message received. name:{type_name}");
353                            match future.await {
354                                Ok(_) => {
355                                    log::debug!("ssh message handled. name:{type_name}");
356                                }
357                                Err(error) => {
358                                    log::error!(
359                                        "error handling message. type:{type_name}, error:{error}",
360                                    );
361                                }
362                            }
363                        } else {
364                            log::error!("unhandled ssh message name:{type_name}");
365                        }
366                    }
367                }
368                anyhow::Ok(())
369            }
370        })
371        .detach();
372
373        this
374    }
375
376    pub fn request<T: RequestMessage>(
377        &self,
378        payload: T,
379    ) -> impl 'static + Future<Output = Result<T::Response>> {
380        log::debug!("ssh request start. name:{}", T::NAME);
381        let response = self.request_dynamic(payload.into_envelope(0, None, None), T::NAME);
382        async move {
383            let response = response.await?;
384            log::debug!("ssh request finish. name:{}", T::NAME);
385            T::Response::from_envelope(response)
386                .ok_or_else(|| anyhow!("received a response of the wrong type"))
387        }
388    }
389
390    pub fn send<T: EnvelopedMessage>(&self, payload: T) -> Result<()> {
391        log::debug!("ssh send name:{}", T::NAME);
392        self.send_dynamic(payload.into_envelope(0, None, None))
393    }
394
395    pub fn request_dynamic(
396        &self,
397        mut envelope: proto::Envelope,
398        type_name: &'static str,
399    ) -> impl 'static + Future<Output = Result<proto::Envelope>> {
400        envelope.id = self.next_message_id.fetch_add(1, SeqCst);
401        let (tx, rx) = oneshot::channel();
402        let mut response_channels_lock = self.response_channels.lock();
403        response_channels_lock.insert(MessageId(envelope.id), tx);
404        drop(response_channels_lock);
405        self.outgoing_tx.unbounded_send(envelope).ok();
406        async move {
407            let response = rx.await.context("connection lost")?.0;
408            if let Some(proto::envelope::Payload::Error(error)) = &response.payload {
409                return Err(RpcError::from_proto(error, type_name));
410            }
411            Ok(response)
412        }
413    }
414
415    pub fn send_dynamic(&self, mut envelope: proto::Envelope) -> Result<()> {
416        envelope.id = self.next_message_id.fetch_add(1, SeqCst);
417        self.outgoing_tx.unbounded_send(envelope)?;
418        Ok(())
419    }
420
421    pub fn subscribe_to_entity<E: 'static>(&self, remote_id: u64, entity: &Model<E>) {
422        let id = (TypeId::of::<E>(), remote_id);
423
424        let mut state = self.state.lock();
425        if state.entities_by_type_and_remote_id.contains_key(&id) {
426            panic!("already subscribed to entity");
427        }
428
429        state.entities_by_type_and_remote_id.insert(
430            id,
431            EntityMessageSubscriber::Entity {
432                handle: entity.downgrade().into(),
433            },
434        );
435    }
436
437    pub async fn spawn_process(&self, command: String) -> process::Child {
438        let (process_tx, process_rx) = oneshot::channel();
439        self.spawn_process_tx
440            .unbounded_send(SpawnRequest {
441                command,
442                process_tx,
443            })
444            .ok();
445        process_rx.await.unwrap()
446    }
447
448    pub fn ssh_args(&self) -> Vec<String> {
449        self.client_socket.as_ref().unwrap().ssh_args()
450    }
451}
452
453impl ProtoClient for SshSession {
454    fn request(
455        &self,
456        envelope: proto::Envelope,
457        request_type: &'static str,
458    ) -> BoxFuture<'static, Result<proto::Envelope>> {
459        self.request_dynamic(envelope, request_type).boxed()
460    }
461
462    fn send(&self, envelope: proto::Envelope, _message_type: &'static str) -> Result<()> {
463        self.send_dynamic(envelope)
464    }
465
466    fn send_response(&self, envelope: Envelope, _message_type: &'static str) -> anyhow::Result<()> {
467        self.send_dynamic(envelope)
468    }
469
470    fn message_handler_set(&self) -> &Mutex<ProtoMessageHandlerSet> {
471        &self.state
472    }
473
474    fn is_via_collab(&self) -> bool {
475        false
476    }
477}
478
479impl SshClientState {
480    #[cfg(not(unix))]
481    async fn new(
482        _connection_options: SshConnectionOptions,
483        _delegate: Arc<dyn SshClientDelegate>,
484        _cx: &mut AsyncAppContext,
485    ) -> Result<Self> {
486        Err(anyhow!("ssh is not supported on this platform"))
487    }
488
489    #[cfg(unix)]
490    async fn new(
491        connection_options: SshConnectionOptions,
492        delegate: Arc<dyn SshClientDelegate>,
493        cx: &mut AsyncAppContext,
494    ) -> Result<Self> {
495        use futures::{io::BufReader, AsyncBufReadExt as _};
496        use smol::{fs::unix::PermissionsExt as _, net::unix::UnixListener};
497        use util::ResultExt as _;
498
499        delegate.set_status(Some("connecting"), cx);
500
501        let url = connection_options.ssh_url();
502        let temp_dir = tempfile::Builder::new()
503            .prefix("zed-ssh-session")
504            .tempdir()?;
505
506        // Create a domain socket listener to handle requests from the askpass program.
507        let askpass_socket = temp_dir.path().join("askpass.sock");
508        let listener =
509            UnixListener::bind(&askpass_socket).context("failed to create askpass socket")?;
510
511        let askpass_task = cx.spawn(|mut cx| async move {
512            while let Ok((mut stream, _)) = listener.accept().await {
513                let mut buffer = Vec::new();
514                let mut reader = BufReader::new(&mut stream);
515                if reader.read_until(b'\0', &mut buffer).await.is_err() {
516                    buffer.clear();
517                }
518                let password_prompt = String::from_utf8_lossy(&buffer);
519                if let Some(password) = delegate
520                    .ask_password(password_prompt.to_string(), &mut cx)
521                    .await
522                    .context("failed to get ssh password")
523                    .and_then(|p| p)
524                    .log_err()
525                {
526                    stream.write_all(password.as_bytes()).await.log_err();
527                }
528            }
529        });
530
531        // Create an askpass script that communicates back to this process.
532        let askpass_script = format!(
533            "{shebang}\n{print_args} | nc -U {askpass_socket} 2> /dev/null \n",
534            askpass_socket = askpass_socket.display(),
535            print_args = "printf '%s\\0' \"$@\"",
536            shebang = "#!/bin/sh",
537        );
538        let askpass_script_path = temp_dir.path().join("askpass.sh");
539        fs::write(&askpass_script_path, askpass_script).await?;
540        fs::set_permissions(&askpass_script_path, std::fs::Permissions::from_mode(0o755)).await?;
541
542        // Start the master SSH process, which does not do anything except for establish
543        // the connection and keep it open, allowing other ssh commands to reuse it
544        // via a control socket.
545        let socket_path = temp_dir.path().join("ssh.sock");
546        let mut master_process = process::Command::new("ssh")
547            .stdin(Stdio::null())
548            .stdout(Stdio::piped())
549            .stderr(Stdio::piped())
550            .env("SSH_ASKPASS_REQUIRE", "force")
551            .env("SSH_ASKPASS", &askpass_script_path)
552            .args(["-N", "-o", "ControlMaster=yes", "-o"])
553            .arg(format!("ControlPath={}", socket_path.display()))
554            .arg(&url)
555            .spawn()?;
556
557        // Wait for this ssh process to close its stdout, indicating that authentication
558        // has completed.
559        let stdout = master_process.stdout.as_mut().unwrap();
560        let mut output = Vec::new();
561        stdout.read_to_end(&mut output).await?;
562        drop(askpass_task);
563
564        if master_process.try_status()?.is_some() {
565            output.clear();
566            let mut stderr = master_process.stderr.take().unwrap();
567            stderr.read_to_end(&mut output).await?;
568            Err(anyhow!(
569                "failed to connect: {}",
570                String::from_utf8_lossy(&output)
571            ))?;
572        }
573
574        Ok(Self {
575            socket: SshSocket {
576                connection_options,
577                socket_path,
578            },
579            _master_process: master_process,
580            _temp_dir: temp_dir,
581        })
582    }
583
584    async fn ensure_server_binary(
585        &self,
586        delegate: &Arc<dyn SshClientDelegate>,
587        src_path: &Path,
588        dst_path: &Path,
589        version: SemanticVersion,
590        cx: &mut AsyncAppContext,
591    ) -> Result<()> {
592        let mut dst_path_gz = dst_path.to_path_buf();
593        dst_path_gz.set_extension("gz");
594
595        if let Some(parent) = dst_path.parent() {
596            run_cmd(self.socket.ssh_command("mkdir").arg("-p").arg(parent)).await?;
597        }
598
599        let mut server_binary_exists = false;
600        if cfg!(not(debug_assertions)) {
601            if let Ok(installed_version) =
602                run_cmd(self.socket.ssh_command(dst_path).arg("version")).await
603            {
604                if installed_version.trim() == version.to_string() {
605                    server_binary_exists = true;
606                }
607            }
608        }
609
610        if server_binary_exists {
611            log::info!("remote development server already present",);
612            return Ok(());
613        }
614
615        let src_stat = fs::metadata(src_path).await?;
616        let size = src_stat.len();
617        let server_mode = 0o755;
618
619        let t0 = Instant::now();
620        delegate.set_status(Some("uploading remote development server"), cx);
621        log::info!("uploading remote development server ({}kb)", size / 1024);
622        self.upload_file(src_path, &dst_path_gz)
623            .await
624            .context("failed to upload server binary")?;
625        log::info!("uploaded remote development server in {:?}", t0.elapsed());
626
627        delegate.set_status(Some("extracting remote development server"), cx);
628        run_cmd(
629            self.socket
630                .ssh_command("gunzip")
631                .arg("--force")
632                .arg(&dst_path_gz),
633        )
634        .await?;
635
636        delegate.set_status(Some("unzipping remote development server"), cx);
637        run_cmd(
638            self.socket
639                .ssh_command("chmod")
640                .arg(format!("{:o}", server_mode))
641                .arg(dst_path),
642        )
643        .await?;
644
645        Ok(())
646    }
647
648    async fn query_platform(&self) -> Result<SshPlatform> {
649        let os = run_cmd(self.socket.ssh_command("uname").arg("-s")).await?;
650        let arch = run_cmd(self.socket.ssh_command("uname").arg("-m")).await?;
651
652        let os = match os.trim() {
653            "Darwin" => "macos",
654            "Linux" => "linux",
655            _ => Err(anyhow!("unknown uname os {os:?}"))?,
656        };
657        let arch = if arch.starts_with("arm") || arch.starts_with("aarch64") {
658            "aarch64"
659        } else if arch.starts_with("x86") || arch.starts_with("i686") {
660            "x86_64"
661        } else {
662            Err(anyhow!("unknown uname architecture {arch:?}"))?
663        };
664
665        Ok(SshPlatform { os, arch })
666    }
667
668    async fn upload_file(&self, src_path: &Path, dest_path: &Path) -> Result<()> {
669        let mut command = process::Command::new("scp");
670        let output = self
671            .socket
672            .ssh_options(&mut command)
673            .args(
674                self.socket
675                    .connection_options
676                    .port
677                    .map(|port| vec!["-P".to_string(), port.to_string()])
678                    .unwrap_or_default(),
679            )
680            .arg(src_path)
681            .arg(format!(
682                "{}:{}",
683                self.socket.connection_options.scp_url(),
684                dest_path.display()
685            ))
686            .output()
687            .await?;
688
689        if output.status.success() {
690            Ok(())
691        } else {
692            Err(anyhow!(
693                "failed to upload file {} -> {}: {}",
694                src_path.display(),
695                dest_path.display(),
696                String::from_utf8_lossy(&output.stderr)
697            ))
698        }
699    }
700}
701
702impl SshSocket {
703    fn ssh_command<S: AsRef<OsStr>>(&self, program: S) -> process::Command {
704        let mut command = process::Command::new("ssh");
705        self.ssh_options(&mut command)
706            .arg(self.connection_options.ssh_url())
707            .arg(program);
708        command
709    }
710
711    fn ssh_options<'a>(&self, command: &'a mut process::Command) -> &'a mut process::Command {
712        command
713            .stdin(Stdio::piped())
714            .stdout(Stdio::piped())
715            .stderr(Stdio::piped())
716            .args(["-o", "ControlMaster=no", "-o"])
717            .arg(format!("ControlPath={}", self.socket_path.display()))
718    }
719
720    fn ssh_args(&self) -> Vec<String> {
721        vec![
722            "-o".to_string(),
723            "ControlMaster=no".to_string(),
724            "-o".to_string(),
725            format!("ControlPath={}", self.socket_path.display()),
726            self.connection_options.ssh_url(),
727        ]
728    }
729}
730
731async fn run_cmd(command: &mut process::Command) -> Result<String> {
732    let output = command.output().await?;
733    if output.status.success() {
734        Ok(String::from_utf8_lossy(&output.stdout).to_string())
735    } else {
736        Err(anyhow!(
737            "failed to run command: {}",
738            String::from_utf8_lossy(&output.stderr)
739        ))
740    }
741}