From 35f914e2c32330816320df6d367086218549e2d0 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Thu, 20 Aug 2026 22:35:43 -0700 Subject: [PATCH] feat(remote-ssh): forward remote ports to this machine Starting a server on an SSH host left it unreachable from here: the `PortForwardManager` only recorded forwards in a HashMap and logged "not fully implemented - connections will not be forwarded". Implement local (`-L`) forwarding for real. Each forward binds a listener and splices accepted connections onto a `direct-tcpip` channel, so the remote sshd dials the service and both directions are copied. The remote endpoint is a forward's identity; the local port is only an allocation. A requested local port that is taken is replaced rather than refused, and the port originally asked for comes back with the mapping so the UI can say the address moved instead of silently handing over a different one. Privileged ports fall back the same way. Three failure modes worth naming: - Channels are opened through SSHConnectionManager per connection instead of against a cached handle, so a forward survives an automatic reconnect. That same path would let a forward nobody stopped dial back out to a host the user explicitly disconnected from, so each connection checks liveness first, and disconnect tears down the forwards it carried. - A local Docker workspace has no SSH session to carry a channel. It is rejected when the mapping is added rather than on the first request. - Forwards bind loopback only. Republishing a remote service to the local network stays behind an explicit opt-in. Add remote listening-port discovery (ss, then lsof, then netstat, with a marker line naming the dialect) so users can pick a real port instead of guessing one. Discovery forwards nothing on its own. Expose it through a manual dialog reached from the workspace menu while connected: nothing is ever forwarded automatically. The table keeps the remote port and the local address in separate columns because they routinely differ. Reverse forwarding keeps its entry point but now fails loudly; it needs `tcpip-forward` global requests and a `forwarded-tcpip` handler that the session handler does not carry. --- src/apps/desktop/src/api/app_state.rs | 10 +- src/apps/desktop/src/api/ssh_api.rs | 84 +- src/apps/desktop/src/lib.rs | 5 + .../core/src/service/remote_ssh/mod.rs | 6 +- .../src/remote_ssh/disabled.rs | 53 +- .../src/remote_ssh/manager.rs | 221 +--- .../src/remote_ssh/mod.rs | 20 +- .../src/remote_ssh/port_forward.rs | 1078 +++++++++++++++++ .../src/remote_ssh/types.rs | 133 ++ .../remote_ssh_contracts.rs | 85 +- .../src/app/components/NavPanel/MainNav.tsx | 32 +- .../ssh-remote/PortForwardDialog.scss | 239 ++++ .../ssh-remote/PortForwardDialog.test.tsx | 281 +++++ .../features/ssh-remote/PortForwardDialog.tsx | 448 +++++++ .../src/features/ssh-remote/appearance.ts | 4 + src/web-ui/src/features/ssh-remote/index.ts | 1 + src/web-ui/src/features/ssh-remote/sshApi.ts | 37 + src/web-ui/src/features/ssh-remote/types.ts | 42 + src/web-ui/src/locales/en-US/common.json | 32 + src/web-ui/src/locales/zh-CN/common.json | 32 + src/web-ui/src/locales/zh-TW/common.json | 32 + 21 files changed, 2655 insertions(+), 220 deletions(-) create mode 100644 src/crates/services/services-integrations/src/remote_ssh/port_forward.rs create mode 100644 src/web-ui/src/features/ssh-remote/PortForwardDialog.scss create mode 100644 src/web-ui/src/features/ssh-remote/PortForwardDialog.test.tsx create mode 100644 src/web-ui/src/features/ssh-remote/PortForwardDialog.tsx diff --git a/src/apps/desktop/src/api/app_state.rs b/src/apps/desktop/src/api/app_state.rs index 945fca9010..ad1edb2abf 100644 --- a/src/apps/desktop/src/api/app_state.rs +++ b/src/apps/desktop/src/api/app_state.rs @@ -10,7 +10,8 @@ use bitfun_core::miniapp::{ initialize_global_miniapp_manager, seed_builtin_miniapps, JsWorkerPool, MiniAppManager, }; use bitfun_core::service::remote_ssh::{ - init_remote_workspace_manager, RemoteFileService, RemoteTerminalManager, SSHConnectionManager, + init_remote_workspace_manager, PortForwardManager, RemoteFileService, RemoteTerminalManager, + SSHConnectionManager, }; use bitfun_core::service::{announcement, config, filesystem, mcp, search, token_usage, workspace}; use bitfun_core::util::errors::*; @@ -86,6 +87,10 @@ pub struct AppState { pub ssh_manager: Arc>>, pub remote_file_service: Arc>>, pub remote_terminal_manager: Arc>>, + /// Local (`-L`) port forwards. Unlike the services above this needs no + /// `Option`: it is useful before any connection exists, and its own entry + /// points report a missing SSH manager. + pub port_forward_manager: PortForwardManager, pub remote_workspace: Arc>>, pub active_searches: Arc>>>, /// Cancellation flags for active file transfers (download/upload), keyed by transfer_id. @@ -304,6 +309,8 @@ impl AppState { let manager_for_fs = Arc::new(tokio::sync::RwLock::new(Some(manager_arc.as_ref().clone()))); let fs = RemoteFileService::new(manager_for_fs.clone()); let tm = RemoteTerminalManager::new(manager_arc.as_ref().clone()); + let port_forward_manager = + PortForwardManager::with_ssh_manager(manager_arc.as_ref().clone()); // Clone for storing in AppState let fs_for_state = fs.clone(); @@ -346,6 +353,7 @@ impl AppState { ssh_manager, remote_file_service, remote_terminal_manager, + port_forward_manager, remote_workspace, active_searches: Arc::new(Mutex::new(HashMap::new())), active_transfers: Arc::new(Mutex::new(HashMap::new())), diff --git a/src/apps/desktop/src/api/ssh_api.rs b/src/apps/desktop/src/api/ssh_api.rs index 2d72ed00eb..24223bdaa0 100644 --- a/src/apps/desktop/src/api/ssh_api.rs +++ b/src/apps/desktop/src/api/ssh_api.rs @@ -11,7 +11,8 @@ use crate::api::app_state::SSHServiceError; use crate::startup_trace::DesktopStartupTrace; use crate::AppState; use bitfun_core::service::remote_ssh::{ - ConnectionTestReport, DockerContainerInfo, RemoteTreeNode, SSHAuthMethod, SSHConfigEntry, + list_remote_listening_ports, ConnectionTestReport, DockerContainerInfo, PortForward, + PortForwardRequest, RemoteListeningPort, RemoteTreeNode, SSHAuthMethod, SSHConfigEntry, SSHConfigLookupResult, SSHConnectionConfig, SSHConnectionManager, SSHConnectionResult, SavedConnection, ServerInfo, }; @@ -188,6 +189,13 @@ pub async fn ssh_disconnect( connection_id: String, ) -> Result<(), String> { let manager = state.get_ssh_manager_async().await?; + // Tear the forwards down first. They are listeners whose only meaning is + // this session, and leaving one running would let a later connection on it + // reconnect the host the user just closed. + state + .port_forward_manager + .stop_for_connection(&connection_id) + .await; manager .disconnect(&connection_id) .await @@ -197,6 +205,7 @@ pub async fn ssh_disconnect( #[tauri::command] pub async fn ssh_disconnect_all(state: State<'_, AppState>) -> Result<(), String> { let manager = state.get_ssh_manager_async().await?; + state.port_forward_manager.stop_all().await; manager.disconnect_all().await; Ok(()) } @@ -540,8 +549,7 @@ fn validate_remote_name_for_local_download(name: &str) -> Result<(), String> { fn local_download_name_key(name: &str) -> String { #[cfg(any(windows, target_os = "macos"))] { - name.trim_end_matches(['.', ' ']) - .to_lowercase() + name.trim_end_matches(['.', ' ']).to_lowercase() } #[cfg(not(any(windows, target_os = "macos")))] { @@ -1121,6 +1129,76 @@ pub async fn remote_get_workspace_info( Ok(workspace) } +// === Port Forwarding === + +/// Start a local (`-L`) forward and return the mapping that was established. +/// +/// The returned `localPort` is not always the requested one: an unavailable +/// port is replaced rather than refused, and `requestedLocalPort` carries what +/// was asked for so the UI can say so. +#[tauri::command] +pub async fn ssh_start_port_forward( + state: State<'_, AppState>, + request: PortForwardRequest, +) -> Result { + log::info!( + "ssh_start_port_forward: connection={}, remote={}:{}, requested local port={:?}", + request.connection_id, + request.effective_remote_host(), + request.remote_port, + request.preferred_local_port() + ); + state + .port_forward_manager + .start_local_forward(&request) + .await + .map_err(|error| format!("{error:#}")) +} + +#[tauri::command] +pub async fn ssh_stop_port_forward( + state: State<'_, AppState>, + forward_id: String, +) -> Result<(), String> { + state + .port_forward_manager + .stop_forward(&forward_id) + .await + .map_err(|error| format!("{error:#}")) +} + +/// List forwards, optionally narrowed to one connection. +#[tauri::command] +pub async fn ssh_list_port_forwards( + state: State<'_, AppState>, + connection_id: Option, +) -> Result, String> { + Ok(match connection_id { + Some(connection_id) => { + state + .port_forward_manager + .list_forwards_for_connection(&connection_id) + .await + } + None => state.port_forward_manager.list_forwards().await, + }) +} + +/// List the TCP ports currently accepting connections on the remote host. +/// +/// Discovery only. Nothing is forwarded as a result of calling this; it exists +/// so users can pick a real port instead of guessing one. +#[tauri::command] +pub async fn ssh_list_remote_listening_ports( + state: State<'_, AppState>, + connection_id: String, +) -> Result, String> { + let manager = state.get_ssh_manager_async().await?; + list_remote_listening_ports(&manager, &connection_id) + .await + .map_err(|error| format!("{error:#}")) +} + #[cfg(test)] mod tests { use super::{ diff --git a/src/apps/desktop/src/lib.rs b/src/apps/desktop/src/lib.rs index 04a4fe0f44..6fc4ecaaeb 100644 --- a/src/apps/desktop/src/lib.rs +++ b/src/apps/desktop/src/lib.rs @@ -1889,6 +1889,11 @@ pub async fn run() { api::ssh_api::remote_close_workspace, api::ssh_api::remote_remove_workspace, api::ssh_api::remote_get_workspace_info, + // Remote port forwarding (local `-L` mappings, user-driven) + api::ssh_api::ssh_start_port_forward, + api::ssh_api::ssh_stop_port_forward, + api::ssh_api::ssh_list_port_forwards, + api::ssh_api::ssh_list_remote_listening_ports, // Detached task dispatch (controller-side SSH transport) api::dispatch_api::dispatch_list_targets, api::dispatch_api::dispatch_probe_target, diff --git a/src/crates/assembly/core/src/service/remote_ssh/mod.rs b/src/crates/assembly/core/src/service/remote_ssh/mod.rs index 13f4d7f365..5e63ef1882 100644 --- a/src/crates/assembly/core/src/service/remote_ssh/mod.rs +++ b/src/crates/assembly/core/src/service/remote_ssh/mod.rs @@ -20,8 +20,8 @@ pub use bitfun_services_integrations::remote_ssh::{ pub use bitfun_services_integrations::remote_ssh::{dispatch_ssh, relay_deploy}; #[cfg(not(feature = "ssh-remote"))] pub use bitfun_services_integrations::remote_ssh::{ - get_global_remote_exec_process_manager, KnownHostEntry, PTYSession, PortForward, - PortForwardDirection, PortForwardManager, RemoteExecCommandRequest, RemoteExecCommandResponse, + get_global_remote_exec_process_manager, list_remote_listening_ports, KnownHostEntry, + PTYSession, PortForwardManager, RemoteExecCommandRequest, RemoteExecCommandResponse, RemoteExecControlAction, RemoteExecControlOrigin, RemoteExecControlRequest, RemoteExecError, RemoteExecProcessLifecycleEvent, RemoteExecProcessLifecycleStatus, RemoteExecProcessManager, RemoteExecResult, RemoteExecSessionCompletion, RemoteExecSessionCompletionSource, @@ -39,7 +39,7 @@ pub use bitfun_services_integrations::remote_ssh::{ }; #[cfg(feature = "ssh-remote")] pub use bitfun_services_integrations::remote_ssh::{ - KnownHostEntry, PTYSession, PortForward, PortForwardDirection, PortForwardManager, + list_remote_listening_ports, KnownHostEntry, PTYSession, PortForwardManager, SSHConnectionManager, }; #[cfg(feature = "ssh-remote")] diff --git a/src/crates/services/services-integrations/src/remote_ssh/disabled.rs b/src/crates/services/services-integrations/src/remote_ssh/disabled.rs index be42334fe4..e94ebe6707 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/disabled.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/disabled.rs @@ -5,9 +5,9 @@ //! stay behind `ssh-remote` and return explicit unsupported errors here. use crate::remote_ssh::types::{ - RemoteDirEntry, RemoteFileEntry, RemoteTreeNode, SSHCommandOptions, SSHCommandResult, - SSHConfigEntry, SSHConfigLookupResult, SSHConnectionConfig, SSHConnectionResult, - SavedConnection, ServerInfo, + PortForward, PortForwardRequest, RemoteDirEntry, RemoteFileEntry, RemoteListeningPort, + RemoteTreeNode, SSHCommandOptions, SSHCommandResult, SSHConfigEntry, SSHConfigLookupResult, + SSHConnectionConfig, SSHConnectionResult, SavedConnection, ServerInfo, }; use std::path::PathBuf; use std::sync::{Arc, OnceLock}; @@ -591,22 +591,6 @@ impl PTYSession { } } -#[derive(Debug, Clone)] -pub struct PortForward { - pub id: String, - pub local_port: u16, - pub remote_host: String, - pub remote_port: u16, - pub direction: PortForwardDirection, -} - -#[derive(Debug, Clone, Copy, PartialEq)] -pub enum PortForwardDirection { - Local, - Remote, - Dynamic, -} - #[derive(Clone, Default)] pub struct PortForwardManager; @@ -623,11 +607,8 @@ impl PortForwardManager { pub async fn start_local_forward( &self, - _connection_id: &str, - _local_port: u16, - _remote_host: &str, - _remote_port: u16, - ) -> anyhow::Result { + _request: &PortForwardRequest, + ) -> anyhow::Result { Err(unsupported()) } @@ -645,17 +626,41 @@ impl PortForwardManager { Err(unsupported()) } + pub async fn stop_for_connection(&self, _connection_id: &str) -> usize { + 0 + } + pub async fn stop_all(&self) {} pub async fn list_forwards(&self) -> Vec { Vec::new() } + pub async fn list_forwards_for_connection(&self, _connection_id: &str) -> Vec { + Vec::new() + } + + pub async fn find_forward( + &self, + _connection_id: &str, + _remote_host: &str, + _remote_port: u16, + ) -> Option { + None + } + pub async fn is_port_forwarded(&self, _port: u16) -> bool { false } } +pub async fn list_remote_listening_ports( + _manager: &SSHConnectionManager, + _connection_id: &str, +) -> anyhow::Result> { + Err(unsupported()) +} + #[derive(Debug, Clone)] pub struct RemoteFileService; diff --git a/src/crates/services/services-integrations/src/remote_ssh/manager.rs b/src/crates/services/services-integrations/src/remote_ssh/manager.rs index cec6b7a9d3..35fb0588d2 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/manager.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/manager.rs @@ -4035,6 +4035,50 @@ impl SSHConnectionManager { Ok(channel) } + /// Open a `direct-tcpip` channel to `host:port` as reached from the remote + /// end, which is the transport local port forwarding runs on. + /// + /// The handle is re-read from the connection map on every call rather than + /// cached by the caller: reconnect installs a *new* handle, and a forward + /// holding the old one would keep splicing traffic into a dead session. + pub async fn open_direct_tcpip( + &self, + connection_id: &str, + host: &str, + port: u16, + originator_host: &str, + originator_port: u16, + ) -> anyhow::Result> { + self.ensure_alive_or_reconnect(connection_id).await?; + let (handle, config) = { + let guard = self.connections.read().await; + let connection = guard + .get(connection_id) + .ok_or_else(|| anyhow!("Connection {} not found", connection_id))?; + ( + connection.handle.clone(), + connection.effective_config.clone(), + ) + }; + if config.uses_local_docker() { + anyhow::bail!("A local Docker workspace has no SSH transport to forward over"); + } + let handle = + handle.ok_or_else(|| anyhow!("SSH handle is unavailable for {}", connection_id))?; + + handle + .channel_open_direct_tcpip(host, port as u32, originator_host, originator_port as u32) + .await + .map_err(|e| { + anyhow!( + "Failed to reach {}:{} from the remote host: {}", + host, + port, + e + ) + }) + } + /// Open a transport-neutral, long-lived stdio process in the effective /// workspace target. /// @@ -5543,183 +5587,6 @@ impl PTYSession { } } -// ============================================================================ -// Port Forwarding -// ============================================================================ - -/// Port forwarding entry -#[derive(Debug, Clone)] -pub struct PortForward { - pub id: String, - pub local_port: u16, - pub remote_host: String, - pub remote_port: u16, - pub direction: PortForwardDirection, -} - -#[derive(Debug, Clone, Copy, PartialEq)] -pub enum PortForwardDirection { - Local, // -L: forward local port to remote - Remote, // -R: forward remote port to local - Dynamic, // -D: dynamic SOCKS proxy -} - -/// Port forwarding manager -pub struct PortForwardManager { - forwards: Arc>>, - ssh_manager: Arc>>, -} - -impl PortForwardManager { - pub fn new() -> Self { - Self { - forwards: Arc::new(tokio::sync::RwLock::new(HashMap::new())), - ssh_manager: Arc::new(tokio::sync::RwLock::new(None)), - } - } - - pub fn with_ssh_manager(ssh_manager: SSHConnectionManager) -> Self { - Self { - forwards: Arc::new(tokio::sync::RwLock::new(HashMap::new())), - ssh_manager: Arc::new(tokio::sync::RwLock::new(Some(ssh_manager))), - } - } - - pub async fn set_ssh_manager(&self, manager: SSHConnectionManager) { - let mut guard = self.ssh_manager.write().await; - *guard = Some(manager); - } - - /// Start local port forwarding (-L) - /// - /// TODO: Full implementation requires: - /// - TCP listener to accept local connections - /// - SSH channel for each forwarded connection - /// - Proper cleanup when stopping the forward - /// - /// Currently this is a placeholder that only tracks the forward configuration. - pub async fn start_local_forward( - &self, - _connection_id: &str, - local_port: u16, - remote_host: String, - remote_port: u16, - ) -> anyhow::Result { - let id = uuid::Uuid::new_v4().to_string(); - - let forward = PortForward { - id: id.clone(), - local_port, - remote_host: remote_host.clone(), - remote_port, - direction: PortForwardDirection::Local, - }; - - // Store forward entry - let mut guard = self.forwards.write().await; - guard.insert(id.clone(), forward); - - log::info!( - "[TODO] Local port forward registered: localhost:{} -> {}:{}", - local_port, - remote_host, - remote_port - ); - log::warn!("Port forwarding is not fully implemented - connections will not be forwarded"); - - Ok(id) - } - - /// Start remote port forwarding (-R) - /// - /// TODO: Full implementation requires SSH reverse port forwarding channel. - /// This is more complex as it needs to bind to a remote port. - pub async fn start_remote_forward( - &self, - _connection_id: &str, - remote_port: u16, - local_host: String, - local_port: u16, - ) -> anyhow::Result { - let id = uuid::Uuid::new_v4().to_string(); - - let forward = PortForward { - id: id.clone(), - local_port: remote_port, - remote_host: local_host.clone(), - remote_port: local_port, - direction: PortForwardDirection::Remote, - }; - - // Remote port forwarding requires SSH channel forwarding - // This is a placeholder - full implementation would need: - // 1. Open a "reverse" channel on SSH connection - // 2. Bind to remote port - // 3. Forward connections back through the channel - - let mut guard = self.forwards.write().await; - guard.insert(id.clone(), forward); - - log::info!( - "Started remote port forward (placeholder): *:{} -> {}:{}", - remote_port, - local_host, - local_port - ); - - // TODO: Implement actual SSH reverse port forwarding - log::warn!("Remote port forwarding is not fully implemented - data will not be forwarded"); - - Ok(id) - } - - /// Stop a port forward - pub async fn stop_forward(&self, forward_id: &str) -> anyhow::Result<()> { - let mut guard = self.forwards.write().await; - if let Some(forward) = guard.remove(forward_id) { - log::info!( - "Stopped port forward: {} ({}:{} -> {}:{})", - forward.id, - match forward.direction { - PortForwardDirection::Local => "local", - PortForwardDirection::Remote => "remote", - PortForwardDirection::Dynamic => "dynamic", - }, - forward.local_port, - forward.remote_host, - forward.remote_port - ); - } - Ok(()) - } - - /// Stop all port forwards - pub async fn stop_all(&self) { - let mut guard = self.forwards.write().await; - let count = guard.len(); - guard.drain(); - log::info!("All {} port forwards stopped", count); - } - - /// List all active forwards - pub async fn list_forwards(&self) -> Vec { - let guard = self.forwards.read().await; - guard.values().cloned().collect() - } - - /// Check if a port is already forwarded - pub async fn is_port_forwarded(&self, port: u16) -> bool { - let guard = self.forwards.read().await; - guard.values().any(|f| f.local_port == port) - } -} - -impl Default for PortForwardManager { - fn default() -> Self { - Self::new() - } -} - fn sftp_mkdir_all_prefixes(path: &str) -> Vec { let is_absolute = path.starts_with('/'); let mut current = String::new(); diff --git a/src/crates/services/services-integrations/src/remote_ssh/mod.rs b/src/crates/services/services-integrations/src/remote_ssh/mod.rs index abe62202b4..251a1da547 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/mod.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/mod.rs @@ -24,6 +24,8 @@ pub mod manager; #[cfg(feature = "remote-ssh-concrete")] mod password_vault; #[cfg(feature = "remote-ssh-concrete")] +mod port_forward; +#[cfg(feature = "remote-ssh-concrete")] pub mod relay_deploy; #[cfg(feature = "remote-ssh-concrete")] mod release_verify; @@ -50,20 +52,20 @@ pub use workspace_services::{remote_workspace_services, RemoteWorkspaceFs, Remot #[cfg(not(feature = "remote-ssh-concrete"))] pub use disabled::{ - dispatch_ssh, get_global_remote_exec_process_manager, KnownHostEntry, PTYSession, PortForward, - PortForwardDirection, PortForwardManager, RemoteExecCommandRequest, RemoteExecCommandResponse, - RemoteExecControlAction, RemoteExecControlOrigin, RemoteExecControlRequest, RemoteExecError, - RemoteExecProcessLifecycleEvent, RemoteExecProcessLifecycleStatus, RemoteExecProcessManager, - RemoteExecResult, RemoteExecSessionCompletion, RemoteExecSessionCompletionSource, + dispatch_ssh, get_global_remote_exec_process_manager, list_remote_listening_ports, + KnownHostEntry, PTYSession, PortForwardManager, RemoteExecCommandRequest, + RemoteExecCommandResponse, RemoteExecControlAction, RemoteExecControlOrigin, + RemoteExecControlRequest, RemoteExecError, RemoteExecProcessLifecycleEvent, + RemoteExecProcessLifecycleStatus, RemoteExecProcessManager, RemoteExecResult, + RemoteExecSessionCompletion, RemoteExecSessionCompletionSource, RemoteExecSessionCompletionStatus, RemoteFileService, RemoteSendStdinRequest, RemoteTerminalManager, RemoteTerminalSession, RemoteWriteStdinRequest, SSHConnectionManager, SessionStatus, }; #[cfg(feature = "remote-ssh-concrete")] -pub use manager::{ - KnownHostEntry, PTYSession, PortForward, PortForwardDirection, PortForwardManager, - SSHConnectionManager, -}; +pub use manager::{KnownHostEntry, PTYSession, SSHConnectionManager}; +#[cfg(feature = "remote-ssh-concrete")] +pub use port_forward::{list_remote_listening_ports, PortForwardManager}; #[cfg(feature = "remote-ssh-concrete")] pub use remote_exec::{ get_global_remote_exec_process_manager, RemoteExecCommandRequest, RemoteExecCommandResponse, diff --git a/src/crates/services/services-integrations/src/remote_ssh/port_forward.rs b/src/crates/services/services-integrations/src/remote_ssh/port_forward.rs new file mode 100644 index 0000000000..b44b2f92e1 --- /dev/null +++ b/src/crates/services/services-integrations/src/remote_ssh/port_forward.rs @@ -0,0 +1,1078 @@ +//! Local (`ssh -L`) port forwarding over an established SSH session. +//! +//! A forward is a listener on this machine whose accepted connections are +//! spliced onto a `direct-tcpip` channel, so the remote sshd dials the service +//! and the two streams are copied in both directions. +//! +//! Two properties drive the design: +//! +//! - **The remote endpoint is the identity, the local port is an allocation.** +//! Users care that "the dev server on the remote's port 3000 is reachable +//! from here"; which local port carries it is a detail. Binding therefore +//! falls back to an arbitrary free port instead of failing, and reports what +//! was asked for so the UI can explain the move. +//! - **A forward must never resurrect a session the user closed.** Channels are +//! opened through [`SSHConnectionManager`] on every connection rather than +//! against a cached handle, which is what keeps a forward working across an +//! automatic reconnect; the liveness check before each open is what stops +//! that same path from silently dialing back out after an explicit +//! disconnect. + +use anyhow::{anyhow, Context}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::Arc; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::Duration; +use tokio_util::sync::CancellationToken; + +use crate::remote_ssh::manager::SSHConnectionManager; +use crate::remote_ssh::types::{ + PortForward, PortForwardDirection, PortForwardRequest, RemoteListeningPort, SSHCommandOptions, +}; + +/// Upper bound on connections a single forward carries at once. +/// +/// A browser opens a handful of sockets per origin, so this sits far above +/// normal use. It exists so one runaway client cannot open SSH channels without +/// limit on a session that is also carrying the editor's own traffic. +const MAX_CONCURRENT_CONNECTIONS: usize = 256; + +/// Originating address reported to the remote when opening a channel. The +/// remote only uses it for logging. +const ORIGINATOR_HOST: &str = "127.0.0.1"; + +/// Pause after a failed `accept` so a persistent error (descriptor exhaustion, +/// most likely) cannot spin the accept loop hot. +const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(50); + +/// How long the remote gets to report its listening sockets. +const LISTENING_PORT_PROBE_TIMEOUT_MS: u64 = 10_000; + +/// Ports never worth offering as forwarding candidates. Port 22 is the +/// session's own transport: it is listening on every host reachable this way, +/// and forwarding it would only ever produce a second path to the SSH server +/// already carrying the forward. +const UNINTERESTING_LISTENING_PORTS: &[u16] = &[22]; + +/// Counters and health for one running forward. +/// +/// Snapshots read these instead of the accept loop pushing updates, so the UI +/// can poll at whatever rate it likes without the data path paying for it. +struct ForwardRuntime { + active_connections: AtomicU32, + total_connections: AtomicU64, + last_error: std::sync::Mutex>, +} + +impl ForwardRuntime { + fn new() -> Self { + Self { + active_connections: AtomicU32::new(0), + total_connections: AtomicU64::new(0), + last_error: std::sync::Mutex::new(None), + } + } + + fn connection_opened(&self) { + self.active_connections.fetch_add(1, Ordering::SeqCst); + self.total_connections.fetch_add(1, Ordering::SeqCst); + } + + fn connection_closed(&self) { + // Saturate rather than wrap. The counter should never reach zero with a + // connection still open, but underflowing it would report four billion + // active connections in the UI and there is no failure worth that. + let _ = + self.active_connections + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |active| { + Some(active.saturating_sub(1)) + }); + } + + fn record_error(&self, message: String) { + if let Ok(mut guard) = self.last_error.lock() { + *guard = Some(message); + } + } + + /// Clear the health signal after a connection completes normally, so a + /// forward started before its remote service was up stops looking broken + /// once the service arrives. + fn clear_error(&self) { + if let Ok(mut guard) = self.last_error.lock() { + *guard = None; + } + } + + fn last_error(&self) -> Option { + self.last_error + .lock() + .ok() + .and_then(|guard| guard.as_ref().cloned()) + } +} + +/// A registered forward: its descriptor, live counters, and shutdown handle. +struct ForwardEntry { + /// Descriptor fields that never change while the forward runs. Counters and + /// `last_error` on this value are stale by construction and are refilled by + /// [`ForwardEntry::snapshot`]. + descriptor: PortForward, + runtime: Arc, + cancel: CancellationToken, +} + +impl ForwardEntry { + fn snapshot(&self) -> PortForward { + let mut forward = self.descriptor.clone(); + forward.active_connections = self.runtime.active_connections.load(Ordering::SeqCst); + forward.total_connections = self.runtime.total_connections.load(Ordering::SeqCst); + forward.last_error = self.runtime.last_error(); + forward + } +} + +/// Whether a failed bind should be retried on an arbitrary port. +/// +/// `AddrInUse` is the everyday case. `PermissionDenied` covers privileged +/// ports, where telling the user to run the editor as root would be a far worse +/// answer than moving the mapping. +fn bind_error_can_fall_back(error: &std::io::Error) -> bool { + matches!( + error.kind(), + std::io::ErrorKind::AddrInUse | std::io::ErrorKind::PermissionDenied + ) +} + +/// Whether a copy error is just the peer hanging up. +/// +/// Closing a browser tab mid-request produces these, and surfacing them would +/// leave a healthy forward permanently flagged as failing. +fn stream_error_is_benign(error: &std::io::Error) -> bool { + matches!( + error.kind(), + std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::NotConnected + | std::io::ErrorKind::UnexpectedEof + ) +} + +/// Bind the listener for a local forward, falling back to an arbitrary free +/// port when the requested one cannot be taken. +/// +/// Returns the listener and, when the requested port was unavailable, the port +/// that was asked for. Callers need that to tell the user the mapping moved +/// rather than quietly handing back a different address. +async fn bind_forward_listener( + host: &str, + preferred: Option, +) -> anyhow::Result<(TcpListener, Option)> { + let Some(port) = preferred else { + let listener = TcpListener::bind((host, 0)) + .await + .with_context(|| format!("Failed to bind {host} for port forwarding"))?; + return Ok((listener, None)); + }; + + match TcpListener::bind((host, port)).await { + Ok(listener) => Ok((listener, None)), + Err(error) if bind_error_can_fall_back(&error) => { + let listener = TcpListener::bind((host, 0)) + .await + .with_context(|| format!("Failed to bind a fallback local port on {host}"))?; + Ok((listener, Some(port))) + } + Err(error) => Err(anyhow!("Failed to bind {host}:{port}: {error}")), + } +} + +/// Port forwarding manager. +/// +/// Owns local (`-L`) forwards. Reverse and dynamic forwards keep their entry +/// point so callers get an explicit error rather than a silent no-op. +#[derive(Clone)] +pub struct PortForwardManager { + forwards: Arc>>, + ssh_manager: Arc>>, +} + +impl PortForwardManager { + pub fn new() -> Self { + Self { + forwards: Arc::new(tokio::sync::RwLock::new(HashMap::new())), + ssh_manager: Arc::new(tokio::sync::RwLock::new(None)), + } + } + + pub fn with_ssh_manager(ssh_manager: SSHConnectionManager) -> Self { + Self { + forwards: Arc::new(tokio::sync::RwLock::new(HashMap::new())), + ssh_manager: Arc::new(tokio::sync::RwLock::new(Some(ssh_manager))), + } + } + + pub async fn set_ssh_manager(&self, manager: SSHConnectionManager) { + let mut guard = self.ssh_manager.write().await; + *guard = Some(manager); + } + + async fn ssh_manager(&self) -> anyhow::Result { + self.ssh_manager + .read() + .await + .clone() + .ok_or_else(|| anyhow!("SSH manager is not initialized")) + } + + /// Start a local (`-L`) forward and return the mapping that was actually + /// established, which is not always the one that was requested. + pub async fn start_local_forward( + &self, + request: &PortForwardRequest, + ) -> anyhow::Result { + let connection_id = request.connection_id.trim(); + if connection_id.is_empty() { + anyhow::bail!("A port forward needs a connection id"); + } + if request.remote_port == 0 { + anyhow::bail!("A port forward needs a remote port"); + } + + let manager = self.ssh_manager().await?; + if !manager.is_connected(connection_id).await { + anyhow::bail!("SSH connection '{connection_id}' is not active"); + } + // Reject the one target that can never work, here rather than on the + // first request: a local Docker workspace is reached by `docker exec` + // and has no SSH session to carry a channel. Discovering that after + // opening a browser would be a worse way to learn it. + if manager.is_local_container_connection(connection_id).await { + anyhow::bail!( + "A local Docker workspace has no SSH transport to forward over. \ + Publish the container port with `docker run -p` instead." + ); + } + + let remote_host = request.effective_remote_host().to_string(); + let local_host = request.effective_local_host(); + + // Hold the registry across the duplicate check, the bind, and the + // insert. Checking under a read lock and inserting later would let two + // concurrent requests for the same endpoint both pass the check and + // both bind, leaving the user with two local addresses that mean the + // same thing. Binding is a syscall, so serializing starts costs + // nothing worth measuring. + let mut guard = self.forwards.write().await; + + if let Some(existing) = guard.values().find(|entry| { + entry.descriptor.connection_id == connection_id + && entry.descriptor.remote_host == remote_host + && entry.descriptor.remote_port == request.remote_port + }) { + anyhow::bail!( + "{}:{} is already forwarded to {}", + remote_host, + request.remote_port, + existing.descriptor.local_address() + ); + } + + let (listener, requested_local_port) = + bind_forward_listener(local_host, request.preferred_local_port()).await?; + let local_port = listener + .local_addr() + .context("Failed to read the bound local address")? + .port(); + + let descriptor = PortForward { + id: uuid::Uuid::new_v4().to_string(), + connection_id: connection_id.to_string(), + direction: PortForwardDirection::Local, + label: request.normalized_label(), + local_host: local_host.to_string(), + local_port, + requested_local_port, + remote_host, + remote_port: request.remote_port, + active_connections: 0, + total_connections: 0, + last_error: None, + }; + + let runtime = Arc::new(ForwardRuntime::new()); + let cancel = CancellationToken::new(); + + log::info!( + "Port forward started: {} -> {}:{} (connection {})", + descriptor.local_address(), + descriptor.remote_host, + descriptor.remote_port, + descriptor.connection_id + ); + if let Some(requested) = requested_local_port { + log::info!( + "Local port {} was unavailable; forward {} bound {} instead", + requested, + descriptor.id, + local_port + ); + } + + Self::spawn_accept_loop( + listener, + manager, + descriptor.clone(), + runtime.clone(), + cancel.clone(), + ); + + guard.insert( + descriptor.id.clone(), + ForwardEntry { + descriptor: descriptor.clone(), + runtime, + cancel, + }, + ); + + Ok(descriptor) + } + + /// Reverse (`-R`) forwarding entry point. + /// + /// Not implemented: it needs `tcpip-forward` global requests and a handler + /// for server-initiated `forwarded-tcpip` channels, neither of which the + /// session handler carries today. It fails loudly rather than registering a + /// forward that moves no data. + pub async fn start_remote_forward( + &self, + _connection_id: &str, + _remote_port: u16, + _local_host: &str, + _local_port: u16, + ) -> anyhow::Result { + Err(anyhow!("Reverse port forwarding is not supported yet")) + } + + fn spawn_accept_loop( + listener: TcpListener, + manager: SSHConnectionManager, + descriptor: PortForward, + runtime: Arc, + cancel: CancellationToken, + ) { + tokio::spawn(async move { + let permits = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_CONNECTIONS)); + loop { + let accepted = tokio::select! { + _ = cancel.cancelled() => break, + accepted = listener.accept() => accepted, + }; + + let (stream, peer) = match accepted { + Ok(accepted) => accepted, + Err(error) => { + // One client failing its handshake must not take the + // whole forward down; report it and keep listening. + runtime + .record_error(format!("Failed to accept a local connection: {error}")); + tokio::time::sleep(ACCEPT_ERROR_BACKOFF).await; + continue; + } + }; + + // Wait for a slot under the same cancellation as the accept + // above. Awaiting the permit bare would keep a stopped forward + // alive until an in-flight connection released one, and then + // open a channel for a client nobody is going to serve. + let permit = tokio::select! { + _ = cancel.cancelled() => break, + permit = permits.clone().acquire_owned() => match permit { + Ok(permit) => permit, + Err(_) => break, + }, + }; + + let manager = manager.clone(); + let descriptor = descriptor.clone(); + let runtime = runtime.clone(); + let cancel = cancel.clone(); + tokio::spawn(async move { + Self::carry_connection( + manager, + descriptor, + runtime, + cancel, + stream, + peer.port(), + ) + .await; + drop(permit); + }); + } + + log::info!( + "Port forward stopped: {} -> {}:{}", + descriptor.local_address(), + descriptor.remote_host, + descriptor.remote_port + ); + }); + } + + async fn carry_connection( + manager: SSHConnectionManager, + descriptor: PortForward, + runtime: Arc, + cancel: CancellationToken, + mut local: TcpStream, + peer_port: u16, + ) { + // Gate on liveness first. `open_direct_tcpip` reconnects on demand, + // which is exactly what keeps a forward alive across a dropped session + // — and exactly what would let a forward the caller forgot to stop dial + // back out to a host the user explicitly disconnected from. + if !manager.is_connected(&descriptor.connection_id).await { + runtime.record_error("SSH connection is not active".to_string()); + return; + } + + let channel = match manager + .open_direct_tcpip( + &descriptor.connection_id, + &descriptor.remote_host, + descriptor.remote_port, + ORIGINATOR_HOST, + peer_port, + ) + .await + { + Ok(channel) => channel, + Err(error) => { + runtime.record_error(format!("{error:#}")); + return; + } + }; + + runtime.connection_opened(); + let mut remote = channel.into_stream(); + let outcome = tokio::select! { + _ = cancel.cancelled() => None, + copied = tokio::io::copy_bidirectional(&mut local, &mut remote) => Some(copied), + }; + match outcome { + Some(Err(error)) if !stream_error_is_benign(&error) => { + runtime.record_error(format!("Forwarded connection failed: {error}")); + } + Some(_) => runtime.clear_error(), + None => {} + } + runtime.connection_closed(); + } + + /// Stop one forward. Idempotent: stopping an unknown id succeeds so a UI + /// that raced a teardown does not have to special-case the reply. + pub async fn stop_forward(&self, forward_id: &str) -> anyhow::Result<()> { + let removed = { + let mut guard = self.forwards.write().await; + guard.remove(forward_id) + }; + if let Some(entry) = removed { + entry.cancel.cancel(); + log::info!( + "Stopped port forward {}: {} -> {}:{}", + entry.descriptor.id, + entry.descriptor.local_address(), + entry.descriptor.remote_host, + entry.descriptor.remote_port + ); + } + Ok(()) + } + + /// Stop every forward carried by one connection. + /// + /// Called when a connection is torn down, so the listeners go away with the + /// session that gave them meaning. + pub async fn stop_for_connection(&self, connection_id: &str) -> usize { + let removed: Vec = { + let mut guard = self.forwards.write().await; + let ids: Vec = guard + .values() + .filter(|entry| entry.descriptor.connection_id == connection_id) + .map(|entry| entry.descriptor.id.clone()) + .collect(); + ids.iter().filter_map(|id| guard.remove(id)).collect() + }; + for entry in &removed { + entry.cancel.cancel(); + } + if !removed.is_empty() { + log::info!( + "Stopped {} port forward(s) for connection {}", + removed.len(), + connection_id + ); + } + removed.len() + } + + /// Stop all port forwards. + pub async fn stop_all(&self) { + let removed: Vec = { + let mut guard = self.forwards.write().await; + guard.drain().map(|(_, entry)| entry).collect() + }; + for entry in &removed { + entry.cancel.cancel(); + } + if !removed.is_empty() { + log::info!("Stopped all {} port forward(s)", removed.len()); + } + } + + /// List every forward, ordered so the UI does not reshuffle between polls. + pub async fn list_forwards(&self) -> Vec { + let guard = self.forwards.read().await; + let mut forwards: Vec = guard.values().map(ForwardEntry::snapshot).collect(); + forwards.sort_by(|left, right| { + left.connection_id + .cmp(&right.connection_id) + .then(left.remote_port.cmp(&right.remote_port)) + .then(left.remote_host.cmp(&right.remote_host)) + }); + forwards + } + + /// List the forwards carried by one connection. + pub async fn list_forwards_for_connection(&self, connection_id: &str) -> Vec { + self.list_forwards() + .await + .into_iter() + .filter(|forward| forward.connection_id == connection_id) + .collect() + } + + /// Look up the forward that already covers a remote endpoint. + pub async fn find_forward( + &self, + connection_id: &str, + remote_host: &str, + remote_port: u16, + ) -> Option { + let guard = self.forwards.read().await; + guard + .values() + .find(|entry| { + entry.descriptor.connection_id == connection_id + && entry.descriptor.remote_host == remote_host + && entry.descriptor.remote_port == remote_port + }) + .map(ForwardEntry::snapshot) + } + + /// Whether some forward is bound to a local port on this machine. + pub async fn is_port_forwarded(&self, port: u16) -> bool { + let guard = self.forwards.read().await; + guard + .values() + .any(|entry| entry.descriptor.local_port == port) + } +} + +impl Default for PortForwardManager { + fn default() -> Self { + Self::new() + } +} + +// ============================================================================ +// Remote listening port discovery +// ============================================================================ + +/// Which tool produced a listening-socket listing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ListingFormat { + /// `ss -H -l -t -n -p`, and `netstat -l -t -n -p`, which agree on the + /// column that holds the local address. + ColumnarSocketStat, + /// `lsof -nP -iTCP -sTCP:LISTEN`. + Lsof, +} + +/// Probe script run on the remote. +/// +/// `ss` first because it is the modern Linux tool, then `lsof` because macOS +/// remotes have it and their `netstat` spells `-p` as "protocol", then +/// `netstat` for older Linux hosts. The marker line tells the parser which +/// dialect came back instead of making it guess from the shape. +const LISTENING_PORT_PROBE_SCRIPT: &str = concat!( + "if command -v ss >/dev/null 2>&1; then ", + "echo __BITFUN_SS__; ss -H -l -t -n -p 2>/dev/null; ", + "elif command -v lsof >/dev/null 2>&1; then ", + "echo __BITFUN_LSOF__; lsof -nP -iTCP -sTCP:LISTEN 2>/dev/null; ", + "elif command -v netstat >/dev/null 2>&1; then ", + "echo __BITFUN_NETSTAT__; netstat -l -t -n -p 2>/dev/null; ", + "else echo __BITFUN_NONE__; fi" +); + +/// Split a `host:port` token, tolerating the IPv6 and wildcard spellings the +/// various tools emit (`[::]:80`, `*:3000`, `:::80`). +fn split_host_port(token: &str) -> Option<(String, u16)> { + let (host, port) = token.rsplit_once(':')?; + let port: u16 = port.trim().parse().ok()?; + let host = host.trim(); + let host = host + .strip_prefix('[') + .and_then(|rest| rest.strip_suffix(']')) + .unwrap_or(host); + let host = match host { + "" | "*" => "0.0.0.0", + other => other, + }; + Some((host.to_string(), port)) +} + +/// Pull the process name and pid out of the trailing column. +/// +/// `ss` writes `users:(("node",pid=999,fd=20))`; `netstat` writes `999/node`. +fn parse_socket_stat_process(line: &str) -> (Option, Option) { + if let Some(users) = line.split("users:((").nth(1) { + let name = users + .split('"') + .nth(1) + .map(str::to_string) + .filter(|name| !name.is_empty()); + let pid = users + .split("pid=") + .nth(1) + .and_then(|rest| { + rest.split(|c: char| !c.is_ascii_digit()) + .find(|part| !part.is_empty()) + }) + .and_then(|digits| digits.parse().ok()); + return (name, pid); + } + + for field in line.split_whitespace() { + if let Some((pid, name)) = field.split_once('/') { + if let Ok(pid) = pid.parse::() { + let name = name.trim(); + return ( + (!name.is_empty() && name != "-").then(|| name.to_string()), + Some(pid), + ); + } + } + } + (None, None) +} + +fn parse_columnar_socket_stat(output: &str) -> Vec { + let mut ports = Vec::new(); + for line in output.lines() { + let fields: Vec<&str> = line.split_whitespace().collect(); + // `ss -H` emits `LISTEN Recv-Q Send-Q Local Peer [users]`; `netstat` + // emits `Proto Recv-Q Send-Q Local Foreign State [pid/name]`. Both put + // the local address in the fourth column, and neither header row + // survives the state/protocol check below. + if fields.len() < 4 { + continue; + } + let is_listening_row = fields[0].eq_ignore_ascii_case("LISTEN") + || (fields[0].starts_with("tcp") + && fields + .iter() + .any(|field| field.eq_ignore_ascii_case("LISTEN"))); + if !is_listening_row { + continue; + } + let Some((bind_address, port)) = split_host_port(fields[3]) else { + continue; + }; + let (process, pid) = parse_socket_stat_process(line); + ports.push(RemoteListeningPort { + port, + bind_address, + process, + pid, + }); + } + ports +} + +fn parse_lsof(output: &str) -> Vec { + let mut ports = Vec::new(); + for line in output.lines() { + if !line.contains("(LISTEN)") { + continue; + } + let fields: Vec<&str> = line.split_whitespace().collect(); + // `COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME (LISTEN)`. + if fields.len() < 9 { + continue; + } + // The name column is the field immediately before `(LISTEN)`, which is + // more robust than indexing from the left: command names can contain + // spaces on some platforms. + let Some(name_index) = fields.iter().position(|field| *field == "(LISTEN)") else { + continue; + }; + let Some(address) = name_index + .checked_sub(1) + .and_then(|index| fields.get(index)) + else { + continue; + }; + // lsof writes `host:port` but also `*:port` and `[::1]:port`. + let Some((bind_address, port)) = split_host_port(address) else { + continue; + }; + ports.push(RemoteListeningPort { + port, + bind_address, + process: Some(fields[0].to_string()).filter(|name| !name.is_empty()), + pid: fields[1].parse().ok(), + }); + } + ports +} + +/// Collapse a listing into one row per port. +/// +/// A service bound to both address families shows up twice, and the port is the +/// only part the user acts on. Rows that name a process win, so the merged +/// entry keeps the most useful label. +fn dedupe_listening_ports(ports: Vec) -> Vec { + let mut merged: Vec = Vec::new(); + for port in ports { + if UNINTERESTING_LISTENING_PORTS.contains(&port.port) { + continue; + } + match merged.iter_mut().find(|kept| kept.port == port.port) { + Some(kept) => { + if kept.process.is_none() && port.process.is_some() { + kept.process = port.process; + kept.pid = port.pid; + kept.bind_address = port.bind_address; + } + } + None => merged.push(port), + } + } + merged.sort_by_key(|port| port.port); + merged +} + +fn parse_listening_ports(output: &str) -> anyhow::Result> { + let mut lines = output.lines(); + let format = loop { + let Some(line) = lines.next() else { + anyhow::bail!("The remote produced no listening-socket listing"); + }; + match line.trim() { + "__BITFUN_SS__" | "__BITFUN_NETSTAT__" => break ListingFormat::ColumnarSocketStat, + "__BITFUN_LSOF__" => break ListingFormat::Lsof, + "__BITFUN_NONE__" => anyhow::bail!( + "The remote host has none of ss, lsof, or netstat, so its listening ports \ + cannot be detected. Enter the port manually." + ), + // Shell profiles love to print banners; skip until the marker. + _ => continue, + } + }; + + let body: String = lines.collect::>().join("\n"); + let ports = match format { + ListingFormat::ColumnarSocketStat => parse_columnar_socket_stat(&body), + ListingFormat::Lsof => parse_lsof(&body), + }; + Ok(dedupe_listening_ports(ports)) +} + +/// List the TCP ports currently accepting connections on the remote host. +/// +/// This is the opt-in alternative to guessing a port number; nothing is +/// forwarded as a result of calling it. +pub async fn list_remote_listening_ports( + manager: &SSHConnectionManager, + connection_id: &str, +) -> anyhow::Result> { + let result = manager + .execute_command_with_options( + connection_id, + LISTENING_PORT_PROBE_SCRIPT, + SSHCommandOptions { + timeout_ms: Some(LISTENING_PORT_PROBE_TIMEOUT_MS), + cancellation_token: None, + }, + ) + .await + .context("Failed to probe the remote host for listening ports")?; + + // A non-zero status is expected: the probe tolerates a tool that partially + // fails, and only a listing with no marker at all is unusable. + parse_listening_ports(&result.stdout) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn splits_ipv4_ipv6_and_wildcard_addresses() { + assert_eq!( + split_host_port("127.0.0.1:3000"), + Some(("127.0.0.1".to_string(), 3000)) + ); + // The IPv6 wildcard keeps its own spelling: it is reported to the user + // as the address the remote service is actually bound to, and claiming + // it is an IPv4 wildcard would be a lie. + assert_eq!(split_host_port("[::]:8080"), Some(("::".to_string(), 8080))); + assert_eq!(split_host_port(":::80"), Some(("::".to_string(), 80))); + // `*` has no address family attached, so loopback-or-anything is the + // only honest reading and `0.0.0.0` is how the other tools spell it. + assert_eq!( + split_host_port("*:5173"), + Some(("0.0.0.0".to_string(), 5173)) + ); + assert_eq!(split_host_port("no-port"), None); + } + + #[test] + fn parses_ss_listing() { + let output = "__BITFUN_SS__\n\ +LISTEN 0 4096 127.0.0.1:3000 0.0.0.0:* users:((\"node\",pid=999,fd=20))\n\ +LISTEN 0 511 0.0.0.0:8080 0.0.0.0:* users:((\"nginx\",pid=1234,fd=6))\n\ +LISTEN 0 128 0.0.0.0:22 0.0.0.0:* users:((\"sshd\",pid=700,fd=3))\n"; + let ports = parse_listening_ports(output).expect("listing should parse"); + assert_eq!(ports.len(), 2, "port 22 is filtered out: {ports:?}"); + assert_eq!(ports[0].port, 3000); + assert_eq!(ports[0].process.as_deref(), Some("node")); + assert_eq!(ports[0].pid, Some(999)); + assert_eq!(ports[1].port, 8080); + assert_eq!(ports[1].process.as_deref(), Some("nginx")); + } + + #[test] + fn parses_netstat_listing() { + let output = "__BITFUN_NETSTAT__\n\ +Active Internet connections (only servers)\n\ +Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name\n\ +tcp 0 0 127.0.0.1:5173 0.0.0.0:* LISTEN 4242/vite\n\ +tcp6 0 0 :::9229 :::* LISTEN -\n"; + let ports = parse_listening_ports(output).expect("listing should parse"); + assert_eq!(ports.len(), 2); + assert_eq!(ports[0].port, 5173); + assert_eq!(ports[0].process.as_deref(), Some("vite")); + assert_eq!(ports[0].pid, Some(4242)); + assert_eq!(ports[1].port, 9229); + assert_eq!(ports[1].process, None); + } + + #[test] + fn parses_lsof_listing() { + let output = "__BITFUN_LSOF__\n\ +COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME\n\ +node 99999 bob 20u IPv4 0x1234567890abcdef 0t0 TCP 127.0.0.1:3000 (LISTEN)\n\ +Python 55555 bob 3u IPv6 0xfedcba0987654321 0t0 TCP *:8000 (LISTEN)\n"; + let ports = parse_listening_ports(output).expect("listing should parse"); + assert_eq!(ports.len(), 2); + assert_eq!(ports[0].port, 3000); + assert_eq!(ports[0].process.as_deref(), Some("node")); + assert_eq!(ports[0].pid, Some(99999)); + assert_eq!(ports[1].port, 8000); + assert_eq!(ports[1].bind_address, "0.0.0.0"); + } + + #[test] + fn skips_shell_banner_before_the_marker() { + let output = "Welcome to Ubuntu 24.04 LTS\n\ +Last login: Tue Aug 19 09:00:00 2026\n\ +__BITFUN_SS__\n\ +LISTEN 0 4096 127.0.0.1:4000 0.0.0.0:* users:((\"api\",pid=12,fd=7))\n"; + let ports = parse_listening_ports(output).expect("listing should parse"); + assert_eq!(ports.len(), 1); + assert_eq!(ports[0].port, 4000); + } + + #[test] + fn reports_a_remote_without_any_probe_tool() { + let error = parse_listening_ports("__BITFUN_NONE__\n") + .expect_err("a remote with no tooling should be an error"); + assert!( + error.to_string().contains("manually"), + "the message should tell the user what to do instead: {error}" + ); + } + + #[test] + fn merges_dual_stack_rows_and_prefers_the_named_process() { + let ports = dedupe_listening_ports(vec![ + RemoteListeningPort { + port: 3000, + bind_address: "0.0.0.0".to_string(), + process: None, + pid: None, + }, + RemoteListeningPort { + port: 3000, + bind_address: "::".to_string(), + process: Some("node".to_string()), + pid: Some(7), + }, + ]); + assert_eq!(ports.len(), 1); + assert_eq!(ports[0].process.as_deref(), Some("node")); + assert_eq!(ports[0].pid, Some(7)); + } + + #[tokio::test] + async fn falls_back_to_a_free_port_when_the_requested_one_is_taken() { + let occupied = TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("the test needs a port to occupy"); + let taken = occupied.local_addr().expect("bound address").port(); + + let (listener, requested) = bind_forward_listener("127.0.0.1", Some(taken)) + .await + .expect("binding should fall back rather than fail"); + let bound = listener.local_addr().expect("bound address").port(); + + assert_eq!( + requested, + Some(taken), + "the requested port must be reported so the UI can explain the move" + ); + assert_ne!(bound, taken); + } + + #[tokio::test] + async fn keeps_the_requested_port_when_it_is_free() { + // Freeing a port and asking for it back is inherently racy: anything + // else on the machine — including the rest of this suite, which binds + // ephemeral ports in parallel — may claim the number in between. Retry + // so a stolen port cannot fail a test about the *un*-stolen path. + const ATTEMPTS: usize = 16; + for _ in 0..ATTEMPTS { + let probe = TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("the test needs a free port number"); + let free = probe.local_addr().expect("bound address").port(); + drop(probe); + + let (listener, requested) = bind_forward_listener("127.0.0.1", Some(free)) + .await + .expect("binding must succeed whether or not the port was taken"); + + if requested.is_none() { + assert_eq!( + listener.local_addr().expect("bound address").port(), + free, + "reporting no move means the requested port is the bound one" + ); + return; + } + // The port was taken between the probe and the bind, so this + // attempt exercised the fallback path instead. That path has its + // own test; try again for a genuinely free port. + assert_eq!(requested, Some(free)); + } + panic!("the requested port was taken on all {ATTEMPTS} attempts"); + } + + #[tokio::test] + async fn refuses_to_forward_without_an_ssh_manager() { + let manager = PortForwardManager::new(); + let error = manager + .start_local_forward(&PortForwardRequest { + connection_id: "conn".to_string(), + remote_port: 3000, + remote_host: None, + local_port: None, + expose_on_lan: false, + label: None, + }) + .await + .expect_err("an uninitialized manager cannot forward"); + assert!(error.to_string().contains("not initialized"), "{error}"); + } + + #[tokio::test] + async fn rejects_a_request_with_no_remote_port() { + let manager = PortForwardManager::new(); + let error = manager + .start_local_forward(&PortForwardRequest { + connection_id: "conn".to_string(), + remote_port: 0, + remote_host: None, + local_port: None, + expose_on_lan: false, + label: None, + }) + .await + .expect_err("port 0 is not a service"); + assert!(error.to_string().contains("remote port"), "{error}"); + } + + #[tokio::test] + async fn stopping_an_unknown_forward_succeeds() { + let manager = PortForwardManager::new(); + manager + .stop_forward("does-not-exist") + .await + .expect("stop should be idempotent"); + assert!(manager.list_forwards().await.is_empty()); + } + + #[test] + fn request_defaults_target_loopback_on_both_ends() { + let request = PortForwardRequest { + connection_id: "conn".to_string(), + remote_port: 3000, + remote_host: Some(" ".to_string()), + local_port: Some(0), + expose_on_lan: false, + label: Some(" ".to_string()), + }; + assert_eq!(request.effective_remote_host(), "127.0.0.1"); + assert_eq!(request.effective_local_host(), "127.0.0.1"); + assert_eq!(request.preferred_local_port(), None); + assert_eq!(request.normalized_label(), None); + } + + #[test] + fn exposing_on_lan_is_the_only_way_to_leave_loopback() { + let request = PortForwardRequest { + connection_id: "conn".to_string(), + remote_port: 3000, + remote_host: None, + local_port: None, + expose_on_lan: true, + label: None, + }; + assert_eq!(request.effective_local_host(), "0.0.0.0"); + } + + #[test] + fn wildcard_binds_are_displayed_as_a_reachable_address() { + let forward = PortForward { + id: "id".to_string(), + connection_id: "conn".to_string(), + direction: PortForwardDirection::Local, + label: None, + local_host: "0.0.0.0".to_string(), + local_port: 3001, + requested_local_port: Some(3000), + remote_host: "127.0.0.1".to_string(), + remote_port: 3000, + active_connections: 0, + total_connections: 0, + last_error: None, + }; + assert_eq!(forward.local_address(), "127.0.0.1:3001"); + } +} diff --git a/src/crates/services/services-integrations/src/remote_ssh/types.rs b/src/crates/services/services-integrations/src/remote_ssh/types.rs index 783fbd7577..ed32593794 100644 --- a/src/crates/services/services-integrations/src/remote_ssh/types.rs +++ b/src/crates/services/services-integrations/src/remote_ssh/types.rs @@ -588,3 +588,136 @@ pub struct SSHConfigLookupResult { /// Config entry if found pub config: Option, } + +/// Direction of an SSH port forward. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum PortForwardDirection { + /// `ssh -L`: a listener on this machine whose connections are carried to + /// the remote end. + Local, + /// `ssh -R`: a listener on the remote whose connections are carried back + /// to this machine. + Remote, + /// `ssh -D`: a local SOCKS proxy. + Dynamic, +} + +/// A user-requested local port forward. +/// +/// Only the remote port is required. Everything else has a default that suits +/// the common case: a dev server bound to the remote's loopback interface, +/// reachable from this machine on the same port number when it happens to be +/// free. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PortForwardRequest { + /// SSH connection that carries the forward. + pub connection_id: String, + /// Port the service listens on, as seen from the remote host. + pub remote_port: u16, + /// Address used by the remote sshd to reach the service. Defaults to + /// loopback, which is what a server bound to `localhost` needs. + #[serde(default)] + pub remote_host: Option, + /// Port to bind on this machine. `None` or `0` allocates a free one. + #[serde(default)] + pub local_port: Option, + /// Bind every interface instead of loopback only. + /// + /// This republishes the remote service to the local network, so it stays + /// opt-in and is never implied by any other field. + #[serde(default)] + pub expose_on_lan: bool, + /// Optional user-facing name for the mapping. + #[serde(default)] + pub label: Option, +} + +impl PortForwardRequest { + /// Address the remote sshd should dial, defaulting to loopback. + pub fn effective_remote_host(&self) -> &str { + match self.remote_host.as_deref().map(str::trim) { + Some(host) if !host.is_empty() => host, + _ => "127.0.0.1", + } + } + + /// Interface to bind on this machine. + pub fn effective_local_host(&self) -> &'static str { + if self.expose_on_lan { + "0.0.0.0" + } else { + "127.0.0.1" + } + } + + /// Requested local port, with `0` normalized to "allocate one for me". + pub fn preferred_local_port(&self) -> Option { + self.local_port.filter(|port| *port != 0) + } + + /// Trimmed label, dropping a blank one. + pub fn normalized_label(&self) -> Option { + self.label + .as_deref() + .map(str::trim) + .filter(|label| !label.is_empty()) + .map(str::to_string) + } +} + +/// A live port forward as reported to the UI. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PortForward { + pub id: String, + pub connection_id: String, + pub direction: PortForwardDirection, + pub label: Option, + /// Interface the listener is bound to on this machine. + pub local_host: String, + /// Port actually bound, which is not always the port that was asked for. + pub local_port: u16, + /// Set only when the requested port was unavailable and a different one was + /// bound instead. The UI needs this to explain the mismatch rather than + /// silently handing back an address the user did not ask for. + pub requested_local_port: Option, + pub remote_host: String, + pub remote_port: u16, + /// Connections currently being carried. + pub active_connections: u32, + /// Connections accepted since the forward started. + pub total_connections: u64, + /// Most recent per-connection failure. Kept as the forward's health signal: + /// a forward stays listening even when the remote service is not up yet. + pub last_error: Option, +} + +impl PortForward { + /// Address to hand a browser, in `host:port` form. + /// + /// A wildcard bind is displayed as loopback because that is the address + /// that actually works on this machine. + pub fn local_address(&self) -> String { + let host = match self.local_host.as_str() { + "0.0.0.0" => "127.0.0.1", + "::" => "[::1]", + other => other, + }; + format!("{}:{}", host, self.local_port) + } +} + +/// A listening TCP port discovered on the remote host. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RemoteListeningPort { + pub port: u16, + /// Address the remote service is bound to, as reported by the remote. + pub bind_address: String, + /// Owning process name when the remote tool exposed it. + pub process: Option, + /// Owning process id when the remote tool exposed it. + pub pid: Option, +} diff --git a/src/crates/services/services-integrations/tests/remote_ssh_contracts/remote_ssh_contracts.rs b/src/crates/services/services-integrations/tests/remote_ssh_contracts/remote_ssh_contracts.rs index 4906fc3e69..3e00837143 100644 --- a/src/crates/services/services-integrations/tests/remote_ssh_contracts/remote_ssh_contracts.rs +++ b/src/crates/services/services-integrations/tests/remote_ssh_contracts/remote_ssh_contracts.rs @@ -6,7 +6,8 @@ use bitfun_services_integrations::remote_ssh::{ sanitize_remote_mirror_path_component, sanitize_ssh_connection_id_for_local_dir, sanitize_ssh_hostname_for_mirror, unresolved_remote_session_storage_dir, unresolved_remote_session_storage_key, workspace_logical_key, workspace_session_identity, - ContainerAccess, ContainerWorkspaceConfig, RemoteWorkspace, RemoteWorkspaceRegistry, + ContainerAccess, ContainerWorkspaceConfig, PortForward, PortForwardDirection, + PortForwardRequest, RemoteListeningPort, RemoteWorkspace, RemoteWorkspaceRegistry, SSHAuthMethod, SSHConnectionConfig, SavedAuthType, SavedConnection, LOCAL_WORKSPACE_SSH_HOST, }; @@ -371,3 +372,85 @@ async fn remote_workspace_registry_preserves_legacy_state_and_clear_contract() { assert!(!registry.has_any().await); assert!(!registry.get_state().await.is_active); } + +#[test] +fn port_forward_request_defaults_keep_a_bare_remote_port_usable() { + // The UI is allowed to send nothing but the connection and the remote port; + // every other field has to default to the safe, common-case answer. + let request: PortForwardRequest = serde_json::from_value(serde_json::json!({ + "connectionId": "conn-1", + "remotePort": 3000 + })) + .expect("a minimal request must deserialize"); + + assert_eq!(request.effective_remote_host(), "127.0.0.1"); + assert_eq!( + request.effective_local_host(), + "127.0.0.1", + "a forward must never reach beyond loopback unless asked to" + ); + assert_eq!(request.preferred_local_port(), None); + assert_eq!(request.normalized_label(), None); + assert!(!request.expose_on_lan); +} + +#[test] +fn port_forward_request_honors_an_explicit_lan_exposure() { + let request: PortForwardRequest = serde_json::from_value(serde_json::json!({ + "connectionId": "conn-1", + "remotePort": 8080, + "remoteHost": "10.0.0.5", + "localPort": 18080, + "exposeOnLan": true, + "label": " staging api " + })) + .expect("a full request must deserialize"); + + assert_eq!(request.effective_remote_host(), "10.0.0.5"); + assert_eq!(request.effective_local_host(), "0.0.0.0"); + assert_eq!(request.preferred_local_port(), Some(18080)); + assert_eq!(request.normalized_label().as_deref(), Some("staging api")); +} + +#[test] +fn port_forward_wire_shape_stays_camel_case() { + let forward = PortForward { + id: "forward-1".to_string(), + connection_id: "conn-1".to_string(), + direction: PortForwardDirection::Local, + label: None, + local_host: "127.0.0.1".to_string(), + local_port: 3001, + requested_local_port: Some(3000), + remote_host: "127.0.0.1".to_string(), + remote_port: 3000, + active_connections: 2, + total_connections: 17, + last_error: None, + }; + + let value = serde_json::to_value(&forward).expect("a forward must serialize"); + assert_eq!(value["localPort"], 3001); + assert_eq!( + value["requestedLocalPort"], 3000, + "the UI needs the requested port to explain a moved mapping" + ); + assert_eq!(value["remotePort"], 3000); + assert_eq!(value["activeConnections"], 2); + assert_eq!(value["direction"], "local"); + assert_eq!(forward.local_address(), "127.0.0.1:3001"); +} + +#[test] +fn remote_listening_port_wire_shape_stays_camel_case() { + let port = RemoteListeningPort { + port: 5173, + bind_address: "127.0.0.1".to_string(), + process: Some("vite".to_string()), + pid: Some(4242), + }; + let value = serde_json::to_value(&port).expect("a listening port must serialize"); + assert_eq!(value["bindAddress"], "127.0.0.1"); + assert_eq!(value["process"], "vite"); + assert_eq!(value["pid"], 4242); +} diff --git a/src/web-ui/src/app/components/NavPanel/MainNav.tsx b/src/web-ui/src/app/components/NavPanel/MainNav.tsx index 3c6e7769aa..1504e627b0 100644 --- a/src/web-ui/src/app/components/NavPanel/MainNav.tsx +++ b/src/web-ui/src/app/components/NavPanel/MainNav.tsx @@ -14,7 +14,7 @@ import React, { useCallback, useState, useMemo, useEffect, useRef } from 'react'; import { createPortal } from 'react-dom'; import { getAppearanceOverlayHost } from '@/infrastructure/appearance/runtime/AppearanceOverlayHost'; -import { Plus, FolderOpen, FolderPlus, History, Check, User, Users, Puzzle, Blocks, CalendarClock, ChevronDown, Search } from 'lucide-react'; +import { Plus, FolderOpen, FolderPlus, History, Check, User, Users, Puzzle, Blocks, CalendarClock, ChevronDown, Search, Network } from 'lucide-react'; // import { PanelsTopLeft } from 'lucide-react'; // temporarily hidden: Pages nav entry import { Tooltip } from '@/component-library'; import { useApp } from '../../hooks/useApp'; @@ -45,7 +45,7 @@ import { } from '@/app/utils/projectSessionWorkspace'; import { getRecentWorkspaceLineParts } from '@/shared/utils/recentWorkspaceDisplay'; import { computeFixedPopoverPosition } from '@/shared/utils/fixedPopoverViewport'; -import { useSSHRemoteContext, SSHConnectionDialog, RemoteFileBrowser } from '@/features/ssh-remote'; +import { useSSHRemoteContext, SSHConnectionDialog, RemoteFileBrowser, PortForwardDialog } from '@/features/ssh-remote'; import { useSessionModeStore } from '../../stores/sessionModeStore'; import NavSearchDialog from './NavSearchDialog'; import { useShortcut } from '@/infrastructure/hooks/useShortcut'; @@ -68,6 +68,7 @@ const MainNav: React.FC = ({ }) => { const sshRemote = useSSHRemoteContext(); const [isSSHConnectionDialogOpen, setIsSSHConnectionDialogOpen] = useState(false); + const [isPortForwardDialogOpen, setIsPortForwardDialogOpen] = useState(false); useEffect(() => { if (sshRemote.showFileBrowser) { @@ -328,6 +329,11 @@ const MainNav: React.FC = ({ setIsSSHConnectionDialogOpen(true); }, [closeWorkspaceMenu]); + const handleOpenPortForward = useCallback(() => { + closeWorkspaceMenu(); + setIsPortForwardDialogOpen(true); + }, [closeWorkspaceMenu]); + const handleSelectRemoteWorkspace = useCallback(async (path: string) => { try { await sshRemote.openWorkspace(path); @@ -465,6 +471,20 @@ const MainNav: React.FC = ({ {t('ssh.remote.connect')} + {sshRemote.isConnected && sshRemote.connectionId && ( + + )}