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 && ( + + )}