use crate::{    json_log::LogRecord,    protocol::{        message_len_from_buffer, read_message_with_len, write_message, MessageId, MESSAGE_LEN_SIZE,    },    proxy::ProxyLaunchError,};use anyhow::{anyhow, Context as _, Result};use async_trait::async_trait;use collections::HashMap;use futures::{    channel::{        mpsc::{self, Sender, UnboundedReceiver, UnboundedSender},        oneshot,    },    future::BoxFuture,    select, select_biased, AsyncReadExt as _, Future, FutureExt as _, StreamExt as _,};use gpui::{    AppContext, AsyncAppContext, Context, EventEmitter, Model, ModelContext, SemanticVersion, Task,    WeakModel,};use parking_lot::Mutex;use rpc::{    proto::{self, build_typed_envelope, Envelope, EnvelopedMessage, PeerId, RequestMessage},    AnyProtoClient, EntityMessageSubscriber, ProtoClient, ProtoMessageHandlerSet, RpcError,};use smol::{    fs,    process::{self, Child, Stdio},};use std::{    any::TypeId,    collections::VecDeque,    ffi::OsStr,    fmt,    ops::ControlFlow,    path::{Path, PathBuf},    sync::{        atomic::{AtomicU32, Ordering::SeqCst},        Arc, Weak,    },    time::{Duration, Instant},};use tempfile::TempDir;use util::ResultExt; #[derive(    Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy, serde::Serialize, serde::Deserialize,)]pub struct SshProjectId(pub u64); #[derive(Clone)]pub struct SshSocket {    connection_options: SshConnectionOptions,    socket_path: PathBuf,} #[derive(Debug, Default, Clone, PartialEq, Eq)]pub struct SshConnectionOptions {    pub host: String,    pub username: Option<String>,    pub port: Option<u16>,    pub password: Option<String>,    pub args: Option<Vec<String>>,} impl SshConnectionOptions {    pub fn parse_command_line(input: &str) -> Result<Self> {        let input = input.trim_start_matches("ssh ");        let mut hostname: Option<String> = None;        let mut username: Option<String> = None;        let mut port: Option<u16> = None;        let mut args = Vec::new();         // disallowed: -E, -e, -F, -f, -G, -g, -M, -N, -n, -O, -q, -S, -s, -T, -t, -V, -v, -W        const ALLOWED_OPTS: &[&str] = &[            "-4", "-6", "-A", "-a", "-C", "-K", "-k", "-X", "-x", "-Y", "-y",        ];        const ALLOWED_ARGS: &[&str] = &[            "-B", "-b", "-c", "-D", "-I", "-i", "-J", "-L", "-l", "-m", "-o", "-P", "-p", "-R",            "-w",        ];         let mut tokens = shlex::split(input)            .ok_or_else(|| anyhow!("invalid input"))?            .into_iter();         'outer: while let Some(arg) = tokens.next() {            if ALLOWED_OPTS.contains(&(&arg as &str)) {                args.push(arg.to_string());                continue;            }            if arg == "-p" {                port = tokens.next().and_then(|arg| arg.parse().ok());                continue;            } else if let Some(p) = arg.strip_prefix("-p") {                port = p.parse().ok();                continue;            }            if arg == "-l" {                username = tokens.next();                continue;            } else if let Some(l) = arg.strip_prefix("-l") {                username = Some(l.to_string());                continue;            }            for a in ALLOWED_ARGS {                if arg == *a {                    args.push(arg);                    if let Some(next) = tokens.next() {                        args.push(next);                    }                    continue 'outer;                } else if arg.starts_with(a) {                    args.push(arg);                    continue 'outer;                }            }            if arg.starts_with("-") || hostname.is_some() {                anyhow::bail!("unsupported argument: {:?}", arg);            }            let mut input = &arg as &str;            if let Some((u, rest)) = input.split_once('@') {                input = rest;                username = Some(u.to_string());            }            if let Some((rest, p)) = input.split_once(':') {                input = rest;                port = p.parse().ok()            }            hostname = Some(input.to_string())        }         let Some(hostname) = hostname else {            anyhow::bail!("missing hostname");        };         Ok(Self {            host: hostname.to_string(),            username: username.clone(),            port,            password: None,            args: Some(args),        })    }     pub fn ssh_url(&self) -> String {        let mut result = String::from("ssh://");        if let Some(username) = &self.username {            result.push_str(username);            result.push('@');        }        result.push_str(&self.host);        if let Some(port) = self.port {            result.push(':');            result.push_str(&port.to_string());        }        result    }     pub fn additional_args(&self) -> Option<&Vec<String>> {        self.args.as_ref()    }     fn scp_url(&self) -> String {        if let Some(username) = &self.username {            format!("{}@{}", username, self.host)        } else {            self.host.clone()        }    }     pub fn connection_string(&self) -> String {        let host = if let Some(username) = &self.username {            format!("{}@{}", username, self.host)        } else {            self.host.clone()        };        if let Some(port) = &self.port {            format!("{}:{}", host, port)        } else {            host        }    }     // Uniquely identifies dev server projects on a remote host. Needs to be    // stable for the same dev server project.    pub fn remote_server_identifier(&self) -> String {        let mut identifier = format!("dev-server-{:?}", self.host);        if let Some(username) = self.username.as_ref() {            identifier.push('-');            identifier.push_str(&username);        }        identifier    }} #[derive(Copy, Clone, Debug)]pub struct SshPlatform {    pub os: &'static str,    pub arch: &'static str,} impl SshPlatform {    pub fn triple(&self) -> Option<String> {        Some(format!(            "{}-{}",            self.arch,            match self.os {                "linux" => "unknown-linux-gnu",                "macos" => "apple-darwin",                _ => return None,            }        ))    }} pub trait SshClientDelegate: Send + Sync {    fn ask_password(        &self,        prompt: String,        cx: &mut AsyncAppContext,    ) -> oneshot::Receiver<Result<String>>;    fn remote_server_binary_path(        &self,        platform: SshPlatform,        cx: &mut AsyncAppContext,    ) -> Result<PathBuf>;    fn get_server_binary(        &self,        platform: SshPlatform,        cx: &mut AsyncAppContext,    ) -> oneshot::Receiver<Result<(PathBuf, SemanticVersion)>>;    fn set_status(&self, status: Option<&str>, cx: &mut AsyncAppContext);} impl SshSocket {    fn ssh_command<S: AsRef<OsStr>>(&self, program: S) -> process::Command {        let mut command = process::Command::new("ssh");        self.ssh_options(&mut command)            .arg(self.connection_options.ssh_url())            .arg(program);        command    }     fn ssh_options<'a>(&self, command: &'a mut process::Command) -> &'a mut process::Command {        command            .stdin(Stdio::piped())            .stdout(Stdio::piped())            .stderr(Stdio::piped())            .args(["-o", "ControlMaster=no", "-o"])            .arg(format!("ControlPath={}", self.socket_path.display()))    }     fn ssh_args(&self) -> Vec<String> {        vec![            "-o".to_string(),            "ControlMaster=no".to_string(),            "-o".to_string(),            format!("ControlPath={}", self.socket_path.display()),            self.connection_options.ssh_url(),        ]    }} async fn run_cmd(command: &mut process::Command) -> Result<String> {    let output = command.output().await?;    if output.status.success() {        Ok(String::from_utf8_lossy(&output.stdout).to_string())    } else {        Err(anyhow!(            "failed to run command: {}",            String::from_utf8_lossy(&output.stderr)        ))    }} const MAX_MISSED_HEARTBEATS: usize = 5;const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(5); const MAX_RECONNECT_ATTEMPTS: usize = 3; enum State {    Connecting,    Connected {        ssh_connection: Box<dyn SshRemoteProcess>,        delegate: Arc<dyn SshClientDelegate>,         multiplex_task: Task<Result<()>>,        heartbeat_task: Task<Result<()>>,    },    HeartbeatMissed {        missed_heartbeats: usize,         ssh_connection: Box<dyn SshRemoteProcess>,        delegate: Arc<dyn SshClientDelegate>,         multiplex_task: Task<Result<()>>,        heartbeat_task: Task<Result<()>>,    },    Reconnecting,    ReconnectFailed {        ssh_connection: Box<dyn SshRemoteProcess>,        delegate: Arc<dyn SshClientDelegate>,         error: anyhow::Error,        attempts: usize,    },    ReconnectExhausted,    ServerNotRunning,} impl fmt::Display for State {    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {        match self {            Self::Connecting => write!(f, "connecting"),            Self::Connected { .. } => write!(f, "connected"),            Self::Reconnecting => write!(f, "reconnecting"),            Self::ReconnectFailed { .. } => write!(f, "reconnect failed"),            Self::ReconnectExhausted => write!(f, "reconnect exhausted"),            Self::HeartbeatMissed { .. } => write!(f, "heartbeat missed"),            Self::ServerNotRunning { .. } => write!(f, "server not running"),        }    }} impl State {    fn ssh_connection(&self) -> Option<&dyn SshRemoteProcess> {        match self {            Self::Connected { ssh_connection, .. } => Some(ssh_connection.as_ref()),            Self::HeartbeatMissed { ssh_connection, .. } => Some(ssh_connection.as_ref()),            Self::ReconnectFailed { ssh_connection, .. } => Some(ssh_connection.as_ref()),            _ => None,        }    }     fn can_reconnect(&self) -> bool {        match self {            Self::Connected { .. }            | Self::HeartbeatMissed { .. }            | Self::ReconnectFailed { .. } => true,            State::Connecting            | State::Reconnecting            | State::ReconnectExhausted            | State::ServerNotRunning => false,        }    }     fn is_reconnect_failed(&self) -> bool {        matches!(self, Self::ReconnectFailed { .. })    }     fn is_reconnect_exhausted(&self) -> bool {        matches!(self, Self::ReconnectExhausted { .. })    }     fn is_server_not_running(&self) -> bool {        matches!(self, Self::ServerNotRunning)    }     fn is_reconnecting(&self) -> bool {        matches!(self, Self::Reconnecting { .. })    }     fn heartbeat_recovered(self) -> Self {        match self {            Self::HeartbeatMissed {                ssh_connection,                delegate,                multiplex_task,                heartbeat_task,                ..            } => Self::Connected {                ssh_connection,                delegate,                multiplex_task,                heartbeat_task,            },            _ => self,        }    }     fn heartbeat_missed(self) -> Self {        match self {            Self::Connected {                ssh_connection,                delegate,                multiplex_task,                heartbeat_task,            } => Self::HeartbeatMissed {                missed_heartbeats: 1,                ssh_connection,                delegate,                multiplex_task,                heartbeat_task,            },            Self::HeartbeatMissed {                missed_heartbeats,                ssh_connection,                delegate,                multiplex_task,                heartbeat_task,            } => Self::HeartbeatMissed {                missed_heartbeats: missed_heartbeats + 1,                ssh_connection,                delegate,                multiplex_task,                heartbeat_task,            },            _ => self,        }    }} /// The state of the ssh connection.#[derive(Clone, Copy, Debug, PartialEq, Eq)]pub enum ConnectionState {    Connecting,    Connected,    HeartbeatMissed,    Reconnecting,    Disconnected,} impl From<&State> for ConnectionState {    fn from(value: &State) -> Self {        match value {            State::Connecting => Self::Connecting,            State::Connected { .. } => Self::Connected,            State::Reconnecting | State::ReconnectFailed { .. } => Self::Reconnecting,            State::HeartbeatMissed { .. } => Self::HeartbeatMissed,            State::ReconnectExhausted => Self::Disconnected,            State::ServerNotRunning => Self::Disconnected,        }    }} pub struct SshRemoteClient {    client: Arc<ChannelClient>,    unique_identifier: String,    connection_options: SshConnectionOptions,    state: Arc<Mutex<Option<State>>>,} #[derive(Debug)]pub enum SshRemoteEvent {    Disconnected,} impl EventEmitter<SshRemoteEvent> for SshRemoteClient {} impl SshRemoteClient {    pub fn new(        unique_identifier: String,        connection_options: SshConnectionOptions,        cancellation: oneshot::Receiver<()>,        delegate: Arc<dyn SshClientDelegate>,        cx: &AppContext,    ) -> Task<Result<Option<Model<Self>>>> {        cx.spawn(|mut cx| async move {            let success = Box::pin(async move {                let (outgoing_tx, outgoing_rx) = mpsc::unbounded::<Envelope>();                let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();                let (connection_activity_tx, connection_activity_rx) = mpsc::channel::<()>(1);                 let client =                    cx.update(|cx| ChannelClient::new(incoming_rx, outgoing_tx, cx, "client"))?;                let this = cx.new_model(|_| Self {                    client: client.clone(),                    unique_identifier: unique_identifier.clone(),                    connection_options: connection_options.clone(),                    state: Arc::new(Mutex::new(Some(State::Connecting))),                })?;                 let (ssh_connection, io_task) = Self::establish_connection(                    unique_identifier,                    false,                    connection_options,                    incoming_tx,                    outgoing_rx,                    connection_activity_tx,                    delegate.clone(),                    &mut cx,                )                .await?;                 let multiplex_task = Self::monitor(this.downgrade(), io_task, &cx);                 if let Err(error) = client.ping(HEARTBEAT_TIMEOUT).await {                    log::error!("failed to establish connection: {}", error);                    return Err(error);                }                 let heartbeat_task =                    Self::heartbeat(this.downgrade(), connection_activity_rx, &mut cx);                 this.update(&mut cx, |this, _| {                    *this.state.lock() = Some(State::Connected {

Lines 1–500 of 1910. Load more