feat: 拆分终端与会话基础能力 (#105)

This commit is contained in:
Realm
2026-08-21 14:24:17 +08:00
committed by GitHub
parent 44348d4d75
commit 350dcbc5b4
17 changed files with 3741 additions and 683 deletions
Generated
+25 -27
View File
@@ -324,7 +324,9 @@ dependencies = [
"base64",
"chacha20poly1305",
"chrono",
"csv",
"directories",
"encoding_rs",
"flate2",
"futures",
"gpui",
@@ -354,6 +356,7 @@ dependencies = [
"sys-locale",
"sysinfo 0.33.1",
"tar",
"tempfile",
"thiserror 1.0.69",
"time",
"tokio",
@@ -878,8 +881,7 @@ dependencies = [
[[package]]
name = "block"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d8c1fef690941d3e7788d328517591fecc684c084084702d6ff1641e993699a"
source = "git+https://github.com/Dicklesworthstone/rust-block?rev=b39ae859d1ee8e8cb5eef6a516471f1578d26b96#b39ae859d1ee8e8cb5eef6a516471f1578d26b96"
[[package]]
name = "block-buffer"
@@ -1591,6 +1593,27 @@ dependencies = [
"typenum",
]
[[package]]
name = "csv"
version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "52cd9d68cf7efc6ddfaaee42e7288d3a99d613d4b50f76ce9827ae0c6e14f938"
dependencies = [
"csv-core",
"itoa",
"ryu",
"serde_core",
]
[[package]]
name = "csv-core"
version = "0.1.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "704a3c26996a80471189265814dbc2c257598b96b8a7feae2d31ace646bb9782"
dependencies = [
"memchr",
]
[[package]]
name = "ctor"
version = "1.0.7"
@@ -5561,28 +5584,6 @@ dependencies = [
"toml_edit 0.25.12+spec-1.1.0",
]
[[package]]
name = "proc-macro-error-attr2"
version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96de42df36bb9bba5542fe9f1a054b8cc87e172759a1868aa05c1f3acc89dfc5"
dependencies = [
"proc-macro2",
"quote",
]
[[package]]
name = "proc-macro-error2"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "11ec05c52be0a07b08061f7dd003e7d7092e0472bc731b4af7bb1ef876109802"
dependencies = [
"proc-macro-error-attr2",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "proc-macro2"
version = "1.0.106"
@@ -7292,10 +7293,7 @@ dependencies = [
[[package]]
name = "stacksafe-macro"
version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "172175341049678163e979d9107ca3508046d4d2a7c6682bee46ac541b17db69"
dependencies = [
"proc-macro-error2",
"quote",
"syn",
]
+8
View File
@@ -13,7 +13,9 @@ tokio-socks = "0.5"
alacritty_terminal = { git = "https://github.com/zed-industries/alacritty", rev = "fcf32feacb367b75ec84dd40f041e4fd411d3cc1" }
async-trait = "0.1"
chrono = { version = "0.4", features = ["serde"] }
csv = "1"
directories = "5"
encoding_rs = "0.8"
flate2 = "1"
gpui = { git = "https://github.com/zed-industries/zed" }
gpui_platform = { git = "https://github.com/zed-industries/zed", features = ["font-kit", "runtime_shaders", "wayland", "x11"] }
@@ -32,6 +34,7 @@ serde_json = "1"
ssh-key = "0.6"
sysinfo = "0.33"
tar = "0.4"
tempfile = "3"
thiserror = "1"
tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "io-util", "net", "time", "fs"] }
tracing = "0.1"
@@ -81,3 +84,8 @@ strip = true
[build-dependencies]
winres = "0.1.12"
# Remove these overrides after GPUI moves off block 0.1 and stacksafe 0.1.
[patch.crates-io]
block = { git = "https://github.com/Dicklesworthstone/rust-block", rev = "b39ae859d1ee8e8cb5eef6a516471f1578d26b96" }
stacksafe-macro = { path = "patches/stacksafe-macro" }
+16
View File
@@ -0,0 +1,16 @@
[package]
name = "stacksafe-macro"
version = "0.1.4"
edition = "2021"
rust-version = "1.80.0"
description = "Procedural macro implementation for the stacksafe crate."
license = "Apache-2.0"
repository = "https://github.com/fast/stacksafe"
publish = false
[lib]
proc-macro = true
[dependencies]
quote = "1"
syn = { version = "2", features = ["full"] }
+73
View File
@@ -0,0 +1,73 @@
// Copyright 2025 FastLabs Developers
// Licensed under the Apache License, Version 2.0.
//! Procedural macro implementation for the `stacksafe` crate.
//!
//! This local compatibility patch preserves the 0.1.4 API while replacing
//! the unmaintained `proc-macro-error2` dependency with native `syn` errors.
use proc_macro::TokenStream;
use quote::{quote, ToTokens};
use syn::{parse_macro_input, parse_quote, ItemFn, Path, ReturnType, Type};
#[proc_macro_attribute]
pub fn stacksafe(args: TokenStream, item: TokenStream) -> TokenStream {
let mut crate_path: Option<Path> = None;
let arg_parser = syn::meta::parser(|meta| {
if meta.path.is_ident("crate") {
crate_path = Some(meta.value()?.parse()?);
Ok(())
} else {
Err(meta.error(format!(
"unknown attribute parameter `{}`",
meta.path
.get_ident()
.map_or("unknown".to_string(), |ident| ident.to_string())
)))
}
});
parse_macro_input!(args with arg_parser);
let mut item_fn: ItemFn = match syn::parse(item) {
Ok(item_fn) => item_fn,
Err(error) => {
return syn::Error::new(
error.span(),
"#[stacksafe] can only be applied to functions",
)
.into_compile_error()
.into();
}
};
if let Some(asyncness) = &item_fn.sig.asyncness {
return syn::Error::new_spanned(asyncness, "#[stacksafe] does not support async functions")
.into_compile_error()
.into();
}
let block = item_fn.block;
let ret = match &item_fn.sig.output {
ReturnType::Type(_, ty) if matches!(**ty, Type::ImplTrait(_)) => ReturnType::Default,
_ => item_fn.sig.output.clone(),
};
let stacksafe_crate = crate_path.unwrap_or_else(|| parse_quote!(::stacksafe));
let wrapped_block = quote! {
{
#stacksafe_crate::internal::stacker::maybe_grow(
#stacksafe_crate::get_minimum_stack_size(),
#stacksafe_crate::get_stack_allocation_size(),
#stacksafe_crate::internal::with_protected(move || #ret { #block })
)
}
};
let wrapped_block = match syn::parse2(wrapped_block) {
Ok(block) => block,
Err(error) => return error.into_compile_error().into(),
};
item_fn.block = Box::new(wrapped_block);
item_fn.into_token_stream().into()
}
+84 -13
View File
@@ -1,19 +1,43 @@
use std::{
io::{Read, Write},
sync::mpsc::{self, Sender},
path::Path,
sync::mpsc,
thread,
time::Duration,
};
#[cfg(not(windows))]
use std::time::Instant;
use anyhow::{Context, Result};
use portable_pty::{CommandBuilder, PtySize, native_pty_system};
#[cfg(not(windows))]
use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind};
use crate::terminal::{BackendCommand, BackendEvent, BackendTx};
use crate::terminal::{BackendCommand, BackendEvent, BackendTx, GuardedBackendEventSender};
pub fn spawn_local_terminal(
#[cfg(not(windows))]
const DIRECTORY_POLL_INTERVAL: Duration = Duration::from_secs(1);
#[cfg(not(windows))]
fn local_process_directory(system: &mut System, pid: Pid) -> Option<std::path::PathBuf> {
system.refresh_processes_specifics(
ProcessesToUpdate::Some(&[pid]),
false,
ProcessRefreshKind::nothing().with_cwd(UpdateKind::Always),
);
system
.process(pid)
.and_then(|process| process.cwd())
.map(std::path::PathBuf::from)
}
pub fn spawn_local_terminal_at(
tab_id: String,
cols: u16,
rows: u16,
events: Sender<BackendEvent>,
events: GuardedBackendEventSender,
initial_directory: Option<&Path>,
) -> Result<BackendTx> {
let pty_system = native_pty_system();
let pair = pty_system
@@ -25,15 +49,27 @@ pub fn spawn_local_terminal(
})
.context("open local PTY")?;
let shell = std::env::var("SHELL").unwrap_or_else(|_| {
if cfg!(windows) {
"powershell.exe".into()
} else {
"/bin/zsh".into()
}
});
let shell = if cfg!(windows) {
"powershell.exe".to_string()
} else {
std::env::var("SHELL").unwrap_or_else(|_| "/bin/zsh".into())
};
let mut cmd = CommandBuilder::new(&shell);
#[cfg(windows)]
{
const POWERSHELL_CWD_REPORTER: &str = r#"& {
$global:AshellOriginalPrompt = $function:prompt
function global:prompt {
$promptText = if ($global:AshellOriginalPrompt) { & $global:AshellOriginalPrompt } else { "PS $PWD> " }
$cwd = $PWD.ProviderPath
$encoded = [Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes($cwd))
[Console]::Write("$([char]27)]0;ASHELL_CWD_B64:$encoded$([char]7)")
$promptText
}
}"#;
cmd.args(["-NoLogo", "-NoExit", "-Command", POWERSHELL_CWD_REPORTER]);
}
cmd.env(
"TERM",
std::env::var("TERM").unwrap_or_else(|_| "xterm-256color".into()),
@@ -54,8 +90,13 @@ pub fn spawn_local_terminal(
if let Ok(home) = std::env::var("HOME") {
cmd.env("HOME", home);
}
if let Some(directory) = initial_directory.filter(|path| path.is_dir()) {
cmd.cwd(directory.as_os_str());
}
cmd.env("SHELL", shell);
let mut child = pair.slave.spawn_command(cmd).context("spawn local shell")?;
#[cfg(not(windows))]
let child_pid = child.process_id().map(Pid::from_u32);
drop(pair.slave);
let master = pair.master;
@@ -94,8 +135,15 @@ pub fn spawn_local_terminal(
let write_tab = tab_id.clone();
let write_events = events.clone();
thread::spawn(move || {
#[cfg(not(windows))]
let mut process_system = System::new();
#[cfg(not(windows))]
let mut last_directory = None;
#[cfg(not(windows))]
let mut last_directory_check = None;
loop {
match cmd_rx.recv_timeout(std::time::Duration::from_millis(100)) {
match cmd_rx.recv_timeout(Duration::from_millis(100)) {
Ok(command) => match command {
BackendCommand::Input(bytes) => {
if let Err(err) = writer.write_all(&bytes) {
@@ -116,7 +164,10 @@ pub fn spawn_local_terminal(
});
}
BackendCommand::Close => break,
BackendCommand::SampleMetrics => {}
BackendCommand::SampleMetrics
| BackendCommand::SampleProcesses
| BackendCommand::SamplePorts
| BackendCommand::TerminateProcess { .. } => {}
},
Err(mpsc::RecvTimeoutError::Timeout) => {
if let Ok(Some(status)) = child.try_wait() {
@@ -129,6 +180,26 @@ pub fn spawn_local_terminal(
}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
}
#[cfg(not(windows))]
{
if last_directory_check.is_none_or(|checked_at: Instant| {
checked_at.elapsed() >= DIRECTORY_POLL_INTERVAL
}) {
last_directory_check = Some(Instant::now());
if let Some(directory) =
child_pid.and_then(|pid| local_process_directory(&mut process_system, pid))
{
if last_directory.as_ref() != Some(&directory) {
last_directory = Some(directory.clone());
let _ = write_events.send(BackendEvent::LocalDirectoryChanged {
tab_id: write_tab.clone(),
path: directory,
});
}
}
}
}
}
let _ = child.kill();
});
+33 -23
View File
@@ -1,5 +1,5 @@
use crate::session::config::Session;
use crate::terminal::{BackendCommand, BackendEvent};
use crate::terminal::{BackendCommand, BackendEvent, GuardedBackendEventSender};
use std::io::{Read, Write};
/// Spawn the serial port backend threads.
@@ -8,7 +8,7 @@ pub fn spawn_serial_client(
_handle: &tokio::runtime::Handle,
tab_id: String,
session: Session,
events_tx: std::sync::mpsc::Sender<BackendEvent>,
events_tx: GuardedBackendEventSender,
) -> tokio::sync::mpsc::UnboundedSender<BackendCommand> {
let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::<BackendCommand>();
@@ -91,7 +91,11 @@ pub fn spawn_serial_client(
let _ = port_write.flush();
}
BackendCommand::Close => break,
_ => {}
BackendCommand::Resize { .. }
| BackendCommand::SampleMetrics
| BackendCommand::SampleProcesses
| BackendCommand::SamplePorts
| BackendCommand::TerminateProcess { .. } => {}
}
}
});
@@ -173,24 +177,27 @@ mod tests {
let (events_tx, events_rx) = std::sync::mpsc::channel();
let handle = tokio::runtime::Handle::current();
let session = Session::serial(slave_name, 0);
let cmd_tx = spawn_serial_client(&handle, "test-tab".to_string(), session, events_tx);
let backend_events = GuardedBackendEventSender::new(events_tx);
let cmd_tx = spawn_serial_client(&handle, "test-tab".to_string(), session, backend_events);
// Wait for the Status event
let status_event = events_rx.recv_timeout(std::time::Duration::from_secs(2));
assert!(status_event.is_ok(), "Failed to receive Status event");
if let Ok(BackendEvent::Status { tab_id, .. }) = status_event {
assert_eq!(tab_id, "test-tab");
} else {
panic!("Expected Status event, got: {:?}", status_event);
let status_event = events_rx
.recv_timeout(std::time::Duration::from_secs(2))
.expect("Failed to receive Status event")
.into_current();
match status_event {
Some(BackendEvent::Status { tab_id, .. }) => assert_eq!(tab_id, "test-tab"),
event => panic!("Expected Status event, got: {event:?}"),
}
// Wait for the Connected event
let connected_event = events_rx.recv_timeout(std::time::Duration::from_secs(2));
assert!(connected_event.is_ok(), "Failed to receive Connected event");
if let Ok(BackendEvent::Connected { tab_id }) = connected_event {
assert_eq!(tab_id, "test-tab");
} else {
panic!("Expected Connected event, got: {:?}", connected_event);
let connected_event = events_rx
.recv_timeout(std::time::Duration::from_secs(2))
.expect("Failed to receive Connected event")
.into_current();
match connected_event {
Some(BackendEvent::Connected { tab_id }) => assert_eq!(tab_id, "test-tab"),
event => panic!("Expected Connected event, got: {event:?}"),
}
// 3. Test Reading: Write to PTY master, verify serial backend outputs it to UI
@@ -198,13 +205,16 @@ mod tests {
master_writer.write_all(b"hello serial simulator").unwrap();
master_writer.flush().unwrap();
let output_event = events_rx.recv_timeout(std::time::Duration::from_secs(2));
assert!(output_event.is_ok(), "Failed to receive Output event");
if let Ok(BackendEvent::Output { tab_id, bytes }) = output_event {
assert_eq!(tab_id, "test-tab");
assert_eq!(bytes, b"hello serial simulator");
} else {
panic!("Expected Output event");
let output_event = events_rx
.recv_timeout(std::time::Duration::from_secs(2))
.expect("Failed to receive Output event")
.into_current();
match output_event {
Some(BackendEvent::Output { tab_id, bytes }) => {
assert_eq!(tab_id, "test-tab");
assert_eq!(bytes, b"hello serial simulator");
}
event => panic!("Expected Output event, got: {event:?}"),
}
// 4. Test Writing: Send BackendCommand::Input to backend, verify PTY master reads it
+503 -27
View File
@@ -5,6 +5,7 @@ use std::{
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use base64::Engine as _;
use directories::BaseDirs;
use russh::{
ChannelMsg, Disconnect,
@@ -21,8 +22,11 @@ use crate::{
session_has_explicit_key,
},
},
system::{SystemSnapshot, remote_snapshot_from_kv},
terminal::{BackendCommand, BackendEvent, BackendTx},
system::{
RemotePort, RemoteProcess, SystemSnapshot, remote_ports_from_probe,
remote_processes_from_ps, remote_snapshot_from_kv,
},
terminal::{BackendCommand, BackendEvent, BackendTx, GuardedBackendEventSender},
};
pub fn spawn_ssh_terminal(
@@ -31,7 +35,7 @@ pub fn spawn_ssh_terminal(
session: Session,
cols: u16,
rows: u16,
events: std::sync::mpsc::Sender<BackendEvent>,
events: GuardedBackendEventSender,
) -> BackendTx {
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel::<BackendCommand>();
let task_tab = tab_id.clone();
@@ -58,30 +62,213 @@ pub fn spawn_ssh_terminal(
async fn sample_remote_system_with_handle(
handle: Arc<tokio::sync::Mutex<russh::client::Handle<ClientHandler>>>,
) -> Result<SystemSnapshot> {
let unix_result = async {
let output = execute_remote_command_with_handle(
handle.clone(),
REMOTE_SYSTEM_PROBE,
"Unix remote metrics probe",
)
.await?;
remote_snapshot_from_kv(&output).context("parse Unix remote metrics probe")
}
.await;
match unix_result {
Ok(snapshot) => Ok(snapshot),
Err(unix_error) => {
let command = powershell_encoded_command(REMOTE_WINDOWS_SYSTEM_PROBE);
let output = execute_remote_command_with_handle(
handle,
&command,
"Windows remote metrics probe",
)
.await
.with_context(|| format!("Unix probe failed first: {unix_error:#}"))?;
remote_snapshot_from_kv(&output)
.context("parse Windows remote metrics probe")
.with_context(|| format!("Unix probe failed first: {unix_error:#}"))
}
}
}
async fn execute_remote_command_with_handle(
handle: Arc<tokio::sync::Mutex<russh::client::Handle<ClientHandler>>>,
command: &str,
operation: &str,
) -> Result<String> {
let mut channel = handle
.lock()
.await
.channel_open_session()
.await
.context("open metrics session")?;
.with_context(|| format!("open {operation} session"))?;
channel
.exec(true, REMOTE_SYSTEM_PROBE)
.exec(true, command)
.await
.context("exec remote metrics probe")?;
.with_context(|| format!("execute {operation}"))?;
let mut stdout = Vec::new();
while let Some(msg) = channel.wait().await {
match msg {
ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, ext: _ } => {
stdout.extend_from_slice(&data);
let mut output = Vec::new();
let mut exit_status = None;
let wait_result = tokio::time::timeout(std::time::Duration::from_secs(20), async {
while let Some(msg) = channel.wait().await {
match msg {
ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, ext: _ } => {
output.extend_from_slice(&data);
}
ChannelMsg::ExitStatus {
exit_status: status,
} => exit_status = Some(status),
ChannelMsg::Close => break,
_ => {}
}
ChannelMsg::Close => break,
_ => {}
}
})
.await;
if wait_result.is_err() {
return Err(anyhow!("{operation} timed out after 20 seconds"));
}
let output = String::from_utf8_lossy(&stdout);
remote_snapshot_from_kv(&output)
let output = String::from_utf8_lossy(&output).trim().to_string();
if exit_status.is_none_or(|status| status != 0) {
let detail = if output.is_empty() {
exit_status.map_or_else(
|| "remote command closed without an exit status".to_string(),
|status| format!("exit status {status}"),
)
} else {
output
};
return Err(anyhow!("{operation} failed: {detail}"));
}
Ok(output)
}
async fn sample_remote_processes_with_handle(
handle: Arc<tokio::sync::Mutex<russh::client::Handle<ClientHandler>>>,
) -> Result<Vec<RemoteProcess>> {
let unix_result = async {
let output = execute_remote_command_with_handle(
handle.clone(),
REMOTE_PROCESS_PROBE,
"Unix remote process probe",
)
.await?;
parse_remote_process_probe(&output, "Unix remote process probe")
}
.await;
match unix_result {
Ok(processes) => Ok(processes),
Err(unix_error) => {
let command = powershell_encoded_command(REMOTE_WINDOWS_PROCESS_PROBE);
let output = execute_remote_command_with_handle(
handle,
&command,
"Windows remote process probe",
)
.await
.with_context(|| format!("Unix probe failed first: {unix_error:#}"))?;
parse_remote_process_probe(&output, "Windows remote process probe")
.with_context(|| format!("Unix probe failed first: {unix_error:#}"))
}
}
}
fn parse_remote_process_probe(output: &str, operation: &str) -> Result<Vec<RemoteProcess>> {
let processes = remote_processes_from_ps(output);
if !processes.is_empty() {
return Ok(processes);
}
let detail = output
.lines()
.map(str::trim)
.find(|line| !line.is_empty())
.unwrap_or("empty output");
Err(anyhow!(
"{operation} returned no parseable process rows: {detail}"
))
}
async fn sample_remote_ports_with_handle(
handle: Arc<tokio::sync::Mutex<russh::client::Handle<ClientHandler>>>,
) -> Result<Vec<RemotePort>> {
let unix_result = async {
let output = execute_remote_command_with_handle(
handle.clone(),
REMOTE_PORT_PROBE,
"Unix remote port probe",
)
.await?;
parse_remote_port_probe(&output, "Unix remote port probe")
}
.await;
match unix_result {
Ok(ports) => Ok(ports),
Err(unix_error) => {
let command = powershell_encoded_command(REMOTE_WINDOWS_PORT_PROBE);
let output =
execute_remote_command_with_handle(handle, &command, "Windows remote port probe")
.await
.with_context(|| format!("Unix probe failed first: {unix_error:#}"))?;
parse_remote_port_probe(&output, "Windows remote port probe")
.with_context(|| format!("Unix probe failed first: {unix_error:#}"))
}
}
}
fn parse_remote_port_probe(output: &str, operation: &str) -> Result<Vec<RemotePort>> {
let ports = remote_ports_from_probe(output);
if !ports.is_empty() {
return Ok(ports);
}
if output.trim().is_empty() {
return Ok(Vec::new());
}
let detail = output
.lines()
.map(str::trim)
.find(|line| !line.is_empty())
.unwrap_or("empty output");
Err(anyhow!(
"{operation} returned no parseable port rows: {detail}"
))
}
async fn terminate_remote_process_with_handle(
handle: Arc<tokio::sync::Mutex<russh::client::Handle<ClientHandler>>>,
pid: u32,
) -> Result<()> {
if pid <= 1 {
return Err(anyhow!("refusing to terminate protected PID {pid}"));
}
if let Err(unix_error) = execute_remote_command_with_handle(
handle.clone(),
&format!("kill -TERM {pid}"),
"terminate Unix remote process",
)
.await
{
let command = powershell_encoded_command(&format!(
"$ErrorActionPreference = 'Stop'; Stop-Process -Id {pid} -ErrorAction Stop"
));
execute_remote_command_with_handle(handle, &command, "terminate Windows remote process")
.await
.with_context(|| format!("Unix termination failed first: {unix_error:#}"))?;
}
Ok(())
}
fn powershell_encoded_command(script: &str) -> String {
let utf16_le = script
.encode_utf16()
.flat_map(u16::to_le_bytes)
.collect::<Vec<_>>();
let encoded = base64::engine::general_purpose::STANDARD.encode(utf16_le);
format!("powershell.exe -NoLogo -NoProfile -NonInteractive -EncodedCommand {encoded}")
}
async fn run_ssh(
@@ -90,7 +277,7 @@ async fn run_ssh(
cols: u16,
rows: u16,
mut commands: mpsc::UnboundedReceiver<BackendCommand>,
events: std::sync::mpsc::Sender<BackendEvent>,
events: GuardedBackendEventSender,
) -> Result<()> {
let _ = events.send(BackendEvent::Status {
tab_id: tab_id.clone(),
@@ -162,6 +349,82 @@ async fn run_ssh(
}
});
}
Some(BackendCommand::SampleProcesses) => {
let handle_clone = handle.clone();
let tab_id_clone = tab_id.clone();
let events_clone = events.clone();
tokio::spawn(async move {
match sample_remote_processes_with_handle(handle_clone).await {
Ok(processes) => {
let _ = events_clone.send(BackendEvent::RemoteProcesses {
tab_id: tab_id_clone,
processes,
});
}
Err(err) => {
let _ = events_clone.send(
BackendEvent::RemoteProcessesUnavailable {
tab_id: tab_id_clone,
reason: format!(
"remote process list unavailable: {err:#}"
),
},
);
}
}
});
}
Some(BackendCommand::SamplePorts) => {
let handle_clone = handle.clone();
let tab_id_clone = tab_id.clone();
let events_clone = events.clone();
tokio::spawn(async move {
match sample_remote_ports_with_handle(handle_clone).await {
Ok(ports) => {
let _ = events_clone.send(BackendEvent::RemotePorts {
tab_id: tab_id_clone,
ports,
});
}
Err(err) => {
let _ = events_clone.send(
BackendEvent::RemotePortsUnavailable {
tab_id: tab_id_clone,
reason: format!(
"remote port list unavailable: {err:#}"
),
},
);
}
}
});
}
Some(BackendCommand::TerminateProcess { pid }) => {
let handle_clone = handle.clone();
let tab_id_clone = tab_id.clone();
let events_clone = events.clone();
tokio::spawn(async move {
match terminate_remote_process_with_handle(handle_clone, pid).await {
Ok(()) => {
let _ = events_clone.send(
BackendEvent::RemoteProcessTerminated {
tab_id: tab_id_clone,
pid,
},
);
}
Err(err) => {
let _ = events_clone.send(
BackendEvent::RemoteProcessTerminateFailed {
tab_id: tab_id_clone,
pid,
reason: format!("{err:#}"),
},
);
}
}
});
}
Some(BackendCommand::Close) | None => {
tracing::info!("[ssh] local client closed the session for tab {}", tab_id);
let _ = channel.eof().await;
@@ -222,7 +485,7 @@ async fn run_ssh(
async fn connect_and_authenticate(
tab_id: &str,
session: &Session,
events: &std::sync::mpsc::Sender<BackendEvent>,
events: &GuardedBackendEventSender,
) -> Result<russh::client::Handle<ClientHandler>> {
let config = Arc::new(client::Config {
inactivity_timeout: None,
@@ -306,7 +569,7 @@ async fn connect_and_authenticate(
let passphrase = session.passphrase.trim();
let passphrase = (!passphrase.is_empty()).then_some(passphrase);
let success = if has_explicit_key {
if has_explicit_key {
let keypair = load_session_private_key(session)?;
let algorithm = format!("{:?}", keypair.algorithm());
let _ = events.send(BackendEvent::Status {
@@ -356,8 +619,7 @@ async fn connect_and_authenticate(
));
}
success
};
success
}
}
AuthMethod::Config => {
// SSH Config auth: try the identity file from config, or default keys
@@ -610,15 +872,229 @@ EOF
exit 0
fi
echo "CPU_PERCENT=0.00"
echo "MEM_TOTAL=0"
echo "MEM_USED=0"
echo "SWAP_TOTAL=0"
echo "SWAP_USED=0"
echo "NET_RX=0"
echo "NET_TX=0"
echo "unsupported Unix remote operating system: $os" >&2
exit 2
'"#;
const REMOTE_PROCESS_PROBE: &str = r#"sh -lc '
os=$(uname -s 2>/dev/null || echo unknown)
if [ "$os" = "Linux" ] && [ -r /proc/stat ]; then
before=$(mktemp "${TMPDIR:-/tmp}/ashell-process-before.XXXXXX") || exit 1
after=$(mktemp "${TMPDIR:-/tmp}/ashell-process-after.XXXXXX") || { rm -f "$before"; exit 1; }
trap '"'"'rm -f "$before" "$after"'"'"' EXIT HUP INT TERM
hz=$(getconf CLK_TCK 2>/dev/null || echo 100)
page_size=$(getconf PAGESIZE 2>/dev/null || echo 4096)
capture_processes() {
for process_stat in /proc/[0-9]*/stat; do
[ -r "$process_stat" ] || continue
process_pid=${process_stat#/proc/}
process_pid=${process_pid%/stat}
process_line=$(cat "$process_stat" 2>/dev/null) || continue
process_rest=${process_line##*) }
set -- $process_rest
[ "$#" -ge 22 ] || continue
process_ticks=$((${12} + ${13}))
process_start=${20}
process_rss_pages=${22}
[ "$process_rss_pages" -ge 0 ] 2>/dev/null || process_rss_pages=0
process_memory=$((process_rss_pages * page_size))
process_user=$(stat -c %U "/proc/$process_pid" 2>/dev/null || echo "-")
process_command=$(tr '"'"'\000\t\r\n'"'"' '"'"' '"'"' < "/proc/$process_pid/cmdline" 2>/dev/null)
if [ -z "$process_command" ]; then
process_command=$(tr '"'"'\t\r\n'"'"' '"'"' '"'"' < "/proc/$process_pid/comm" 2>/dev/null)
fi
[ -n "$process_command" ] || process_command="[$process_pid]"
printf '"'"'%s\t%s\t%s\t%s\t%s\t%s\n'"'"' "$process_pid" "$process_user" "$process_ticks" "$process_start" "$process_memory" "$process_command"
done
}
capture_processes > "$before"
sleep 1
capture_processes > "$after"
awk -F '"'"'\t'"'"' -v hz="$hz" '"'"'
NR == FNR { ticks[$1] = $3; starts[$1] = $4; next }
{
delta = ($1 in ticks && starts[$1] == $4) ? ($3 - ticks[$1]) : 0;
cpu = (delta > 0 && hz > 0) ? delta * 100 / hz : 0;
printf "PROCESS\t%s\t%s\t%.2f\t%s\t%s\n", $1, $2, cpu, $5, $6;
}
'"'"' "$before" "$after"
exit 0
fi
if [ "$os" = "Darwin" ]; then
top_output=$(mktemp "${TMPDIR:-/tmp}/ashell-process-top.XXXXXX") || exit 1
trap '"'"'rm -f "$top_output"'"'"' EXIT HUP INT TERM
LC_ALL=C top -l 2 -s 1 -n 10000 -stats pid,cpu > "$top_output" 2>/dev/null || exit 1
LC_ALL=C ps -axo pid=,user=,rss=,command= 2>/dev/null | awk -v top_output="$top_output" '"'"'
BEGIN {
sample = 0;
while ((getline line < top_output) > 0) {
count = split(line, fields, /[[:space:]]+/);
start = fields[1] == "" ? 2 : 1;
if (fields[start] == "PID") { sample++; continue; }
if (sample == 2 && fields[start] ~ /^[0-9]+$/) {
value = fields[start + 1];
gsub(/%/, "", value);
cpu[fields[start]] = value + 0;
}
}
close(top_output);
}
{
pid = $1; user = $2; memory = $3 * 1024;
$1 = $2 = $3 = "";
sub(/^[[:space:]]+/, "");
printf "PROCESS\t%s\t%s\t%.2f\t%.0f\t%s\n", pid, user, cpu[pid] + 0, memory, $0;
}
'"'"'
exit 0
fi
echo "unsupported Unix remote operating system: $os" >&2
exit 2
'"#;
const REMOTE_PORT_PROBE: &str = r#"sh -lc '
if command -v lsof >/dev/null 2>&1; then
lsof -nP -iTCP -sTCP:LISTEN 2>/dev/null | awk '"'"'
NR > 1 && $2 ~ /^[0-9]+$/ {
endpoint = $9;
state = $10;
gsub(/[()]/, "", state);
if (endpoint !~ /:[0-9]+$/) next;
port = endpoint;
sub(/^.*:/, "", port);
address = endpoint;
sub(/:[^:]*$/, "", address);
if (address == "") address = "*";
if (state == "") state = "LISTEN";
printf "PORT\tTCP\t%s\t%s\t%s\t%s\t%s\n", address, port, state, $2, $1;
}
'"'"'
lsof -nP -iUDP 2>/dev/null | awk '"'"'
NR > 1 && $2 ~ /^[0-9]+$/ {
endpoint = $9;
if (endpoint !~ /:[0-9]+$/) next;
port = endpoint;
sub(/^.*:/, "", port);
address = endpoint;
sub(/:[^:]*$/, "", address);
if (address == "") address = "*";
printf "PORT\tUDP\t%s\t%s\tUNCONN\t%s\t%s\n", address, port, $2, $1;
}
'"'"'
exit 0
fi
if command -v ss >/dev/null 2>&1; then
ss -H -lntup 2>/dev/null | awk '"'"'
NF >= 5 {
protocol = $1;
state = $2;
endpoint = $5;
if (endpoint !~ /:[0-9]+$/) next;
port = endpoint;
sub(/^.*:/, "", port);
address = endpoint;
sub(/:[^:]*$/, "", address);
pid = "-";
process = "-";
for (i = 6; i <= NF; i++) {
token = $i;
if (token ~ /users:/) {
name = token;
sub(/^.*\(\("/, "", name);
sub(/".*$/, "", name);
if (name != "") process = name;
if (match(token, /pid=[0-9]+/)) {
pid = substr(token, RSTART + 4, RLENGTH - 4);
}
}
}
printf "PORT\t%s\t%s\t%s\t%s\t%s\t%s\n", protocol, address, port, state, pid, process;
}
'"'"'
exit 0
fi
if command -v netstat >/dev/null 2>&1; then
netstat -an 2>/dev/null | awk '"'"'
NR > 1 && NF >= 4 {
protocol = $1;
endpoint = $4;
state = protocol ~ /^udp/i ? "UNCONN" : "LISTEN";
if (endpoint !~ /:[0-9]+$/ && endpoint !~ /\.[0-9]+$/) next;
port = endpoint;
sub(/^.*[:.]/, "", port);
address = endpoint;
sub(/[:.][^:.]*$/, "", address);
if (address == "") address = "*";
if (protocol !~ /^udp/i && $NF != "LISTEN" && $NF != "LISTENING") next;
printf "PORT\t%s\t%s\t%s\t%s\t-\t-\n", protocol, address, port, state;
}
'"'"'
exit 0
fi
echo "no supported remote port utility found" >&2
exit 2
'"#;
const REMOTE_WINDOWS_SYSTEM_PROBE: &str = r#"$ErrorActionPreference = 'Stop'
[Console]::OutputEncoding = [System.Text.Encoding]::UTF8
$cpu = (Get-CimInstance Win32_Processor | Measure-Object -Property LoadPercentage -Average).Average
$os = Get-CimInstance Win32_OperatingSystem
$memTotal = [uint64]$os.TotalVisibleMemorySize * 1024
$memFree = [uint64]$os.FreePhysicalMemory * 1024
$swapTotal = [uint64]$os.TotalVirtualMemorySize * 1024 - $memTotal
$swapFree = [uint64]$os.FreeVirtualMemory * 1024 - $memFree
$network = Get-CimInstance Win32_PerfFormattedData_Tcpip_NetworkInterface
$netRx = ($network | Measure-Object -Property BytesReceivedPersec -Sum).Sum
$netTx = ($network | Measure-Object -Property BytesSentPersec -Sum).Sum
$cpuText = [string]::Format([Globalization.CultureInfo]::InvariantCulture, "{0:F2}", [double]$cpu)
Write-Output ("CPU_PERCENT={0}" -f $cpuText)
Write-Output ("MEM_TOTAL={0}" -f $memTotal)
Write-Output ("MEM_USED={0}" -f ($memTotal - $memFree))
Write-Output ("SWAP_TOTAL={0}" -f ([Math]::Max(0, $swapTotal)))
Write-Output ("SWAP_USED={0}" -f ([Math]::Max(0, $swapTotal - $swapFree)))
Write-Output ("NET_RX={0}" -f ([uint64]$netRx))
Write-Output ("NET_TX={0}" -f ([uint64]$netTx))
Get-CimInstance Win32_LogicalDisk -Filter "DriveType=3" | ForEach-Object {
Write-Output ("DISK={0}`t{1}`t{2}" -f $_.DeviceID, [uint64]$_.FreeSpace, [uint64]$_.Size)
}"#;
const REMOTE_WINDOWS_PROCESS_PROBE: &str = r#"$ErrorActionPreference = 'Stop'
[Console]::OutputEncoding = [System.Text.Encoding]::UTF8
$before = @{}
Get-Process | ForEach-Object { if ($null -ne $_.CPU) { $before[$_.Id] = [double]$_.CPU } }
Start-Sleep -Seconds 1
Get-Process | ForEach-Object {
$previous = $before[$_.Id]
$cpu = if ($null -ne $previous -and $null -ne $_.CPU) { [Math]::Max(0, ([double]$_.CPU - $previous) * 100) } else { 0 }
$command = $_.ProcessName.Replace("`t", " ").Replace("`r", " ").Replace("`n", " ")
$cpuText = [string]::Format([Globalization.CultureInfo]::InvariantCulture, "{0:F2}", $cpu)
Write-Output ("PROCESS`t{0}`t-`t{1}`t{2}`t{3}" -f $_.Id, $cpuText, [uint64]$_.WorkingSet64, $command)
}"#;
const REMOTE_WINDOWS_PORT_PROBE: &str = r#"$ErrorActionPreference = 'Stop'
[Console]::OutputEncoding = [System.Text.Encoding]::UTF8
$tab = [char]9
$processNames = @{}
Get-Process | ForEach-Object { $processNames[[int]$_.Id] = $_.ProcessName }
@(Get-NetTCPConnection -State Listen -ErrorAction SilentlyContinue) | ForEach-Object {
$ownerPid = [int]$_.OwningProcess
$processName = if ($processNames.ContainsKey($ownerPid)) { $processNames[$ownerPid] } else { '-' }
Write-Output ("PORT{0}TCP{0}{1}{0}{2}{0}LISTEN{0}{3}{0}{4}" -f $tab, $_.LocalAddress, $_.LocalPort, $ownerPid, $processName)
}
@(Get-NetUDPEndpoint -ErrorAction SilentlyContinue) | ForEach-Object {
$ownerPid = [int]$_.OwningProcess
$processName = if ($processNames.ContainsKey($ownerPid)) { $processNames[$ownerPid] } else { '-' }
Write-Output ("PORT{0}UDP{0}{1}{0}{2}{0}UNCONN{0}{3}{0}{4}" -f $tab, $_.LocalAddress, $_.LocalPort, $ownerPid, $processName)
}"#;
#[derive(Clone)]
struct ClientHandler;
+558 -70
View File
@@ -1,4 +1,10 @@
use std::{fs, path::PathBuf, sync::OnceLock};
use std::{
collections::{HashMap, HashSet},
fs,
io::Write as _,
path::{Path, PathBuf},
sync::{Arc, Mutex, OnceLock},
};
use anyhow::{Context, Result};
use argon2::Argon2;
@@ -12,6 +18,8 @@ use rand::{RngCore, rngs::OsRng};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::text_encoding::TextEncoding;
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum AuthMethod {
@@ -60,6 +68,8 @@ pub struct Session {
pub protocol: String,
#[serde(default = "default_baud_rate")]
pub baud_rate: u32,
#[serde(default)]
pub terminal_encoding: TextEncoding,
}
impl Session {
@@ -84,6 +94,7 @@ impl Session {
proxy_password: String::new(),
protocol: "ssh".to_string(),
baud_rate: 115200,
terminal_encoding: TextEncoding::Utf8,
}
}
@@ -115,6 +126,7 @@ impl Session {
proxy_password: String::new(),
protocol: "ssh".to_string(),
baud_rate: 115200,
terminal_encoding: TextEncoding::Utf8,
}
}
@@ -139,6 +151,7 @@ impl Session {
proxy_password: String::new(),
protocol: "serial".to_string(),
baud_rate,
terminal_encoding: TextEncoding::Utf8,
}
}
}
@@ -166,6 +179,61 @@ pub enum SavedWindowBounds {
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum SavedPaneLayout {
Single {
tab_id: String,
},
Horizontal {
children: Vec<SavedPaneLayout>,
ratio: f32,
},
Vertical {
children: Vec<SavedPaneLayout>,
ratio: f32,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum SavedTerminalTab {
Local {
id: String,
#[serde(default)]
cwd: Option<PathBuf>,
#[serde(default)]
terminal_encoding: TextEncoding,
},
Ssh {
id: String,
session: Session,
},
Serial {
id: String,
session: Session,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SavedTabGroup {
pub id: String,
pub title: String,
pub pane_root: SavedPaneLayout,
#[serde(default)]
pub tabs: Vec<SavedTerminalTab>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SavedTabsState {
#[serde(default)]
pub groups: Vec<SavedTabGroup>,
#[serde(default)]
pub active_group: Option<String>,
#[serde(default)]
pub active_tab: Option<String>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "lowercase")]
pub enum TitleBarStyle {
@@ -214,6 +282,15 @@ pub struct ConfigFile {
pub cursor_style: CursorStyle,
#[serde(default)]
pub sessions: Vec<Session>,
/// Shell commands recorded for each SSH session ID.
#[serde(default)]
pub command_history: HashMap<String, Vec<String>>,
#[serde(default)]
pub command_history_revision: u64,
#[serde(default)]
pub remember_tabs: bool,
#[serde(default)]
pub saved_tabs: Option<SavedTabsState>,
#[serde(default)]
pub window_bounds: Option<SavedWindowBounds>,
#[serde(default)]
@@ -221,6 +298,12 @@ pub struct ConfigFile {
#[serde(default)]
pub body_panels: Option<Vec<f32>>,
#[serde(default)]
pub sftp_tree_panels: Option<Vec<f32>>,
#[serde(default)]
pub sftp_file_columns: Option<Vec<f32>>,
#[serde(default)]
pub sftp_file_columns_customized: bool,
#[serde(default)]
pub transfers: Vec<crate::terminal::Transfer>,
#[serde(default)]
pub show_hidden_files: bool,
@@ -316,6 +399,35 @@ fn default_terminal_font_family() -> String {
"Maple Mono NF CN".to_string()
}
const MAX_COMMAND_HISTORY: usize = 200;
fn normalize_command_history_entries(history: &mut Vec<String>) -> bool {
let mut seen = HashSet::new();
let mut changed = false;
let mut normalized = history
.drain(..)
.rev()
.filter_map(|command| {
let trimmed = command.trim();
changed |= trimmed.len() != command.len();
let command = trimmed.to_string();
if command.is_empty() || !seen.insert(command.clone()) {
changed = true;
None
} else {
Some(command)
}
})
.collect::<Vec<_>>();
normalized.reverse();
if normalized.len() > MAX_COMMAND_HISTORY {
changed = true;
normalized.drain(..normalized.len() - MAX_COMMAND_HISTORY);
}
*history = normalized;
changed
}
impl Default for ConfigFile {
fn default() -> Self {
Self {
@@ -333,9 +445,16 @@ impl Default for ConfigFile {
title_bar_style: TitleBarStyle::default(),
cursor_style: CursorStyle::default(),
sessions: Vec::new(),
command_history: HashMap::new(),
command_history_revision: 0,
remember_tabs: false,
saved_tabs: None,
window_bounds: None,
workspace_panels: None,
body_panels: None,
sftp_tree_panels: None,
sftp_file_columns: None,
sftp_file_columns_customized: false,
transfers: Vec::new(),
show_hidden_files: false,
lock_layout: false,
@@ -368,6 +487,80 @@ impl Default for ConfigFile {
pub struct ConfigStore {
pub(crate) path: PathBuf,
pub(crate) cache: ConfigFile,
write_lock: Arc<Mutex<()>>,
}
fn config_backup_path(path: &Path) -> PathBuf {
path.with_extension("json.bak")
}
fn decode_config_bytes(raw_bytes: &[u8], hardware_uuid: &str) -> Result<ConfigFile> {
match decrypt_config(raw_bytes, hardware_uuid) {
Ok(cache) => Ok(cache),
Err(decrypt_err) => serde_json::from_slice::<ConfigFile>(raw_bytes).map_err(|json_err| {
anyhow::anyhow!(
"decrypt failed: {decrypt_err:#}; plain JSON parsing failed: {json_err:#}"
)
}),
}
}
fn persist_config_bytes(path: &Path, contents: &[u8]) -> Result<()> {
let parent = path
.parent()
.context("configuration path has no parent directory")?;
let mut temporary = tempfile::Builder::new()
.prefix(".ashell-config-")
.suffix(".tmp")
.tempfile_in(parent)
.with_context(|| format!("failed to create temporary config in {}", parent.display()))?;
temporary
.write_all(contents)
.with_context(|| format!("failed to write temporary config for {}", path.display()))?;
temporary
.as_file()
.sync_all()
.with_context(|| format!("failed to sync temporary config for {}", path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
let mut permissions = temporary
.as_file()
.metadata()
.with_context(|| format!("failed to inspect temporary config for {}", path.display()))?
.permissions();
permissions.set_mode(0o600);
temporary
.as_file()
.set_permissions(permissions)
.with_context(|| {
format!("failed to protect temporary config for {}", path.display())
})?;
}
temporary
.persist(path)
.map_err(|error| error.error)
.with_context(|| format!("failed to replace {} atomically", path.display()))?;
#[cfg(unix)]
fs::File::open(parent)
.and_then(|directory| directory.sync_all())
.with_context(|| format!("failed to sync config directory {}", parent.display()))?;
Ok(())
}
fn write_config_bytes(path: &Path, contents: &[u8]) -> Result<()> {
if path.exists() {
let previous = fs::read(path)
.with_context(|| format!("failed to read previous config {}", path.display()))?;
persist_config_bytes(&config_backup_path(path), &previous)
.with_context(|| format!("failed to back up config {}", path.display()))?;
}
persist_config_bytes(path, contents)
}
impl ConfigStore {
@@ -395,28 +588,68 @@ impl ConfigStore {
let raw_bytes =
fs::read(&path).with_context(|| format!("failed to read {}", path.display()))?;
let hardware_uuid = get_hardware_uuid();
match decrypt_config(&raw_bytes, &hardware_uuid) {
Ok(cache) => cache,
Err(decrypt_err) => {
// Fallback to plain text JSON if decryption/parsing failed
match serde_json::from_slice::<ConfigFile>(&raw_bytes) {
Ok(cache) => cache,
Err(json_err) => {
let backup_path = path.with_extension("json.bak");
if let Err(backup_err) = fs::write(&backup_path, &raw_bytes) {
tracing::warn!(
"failed to parse config {} (decrypt err: {decrypt_err:#}, json err: {json_err:#}); backup to {} also failed: {backup_err:#}",
match decode_config_bytes(&raw_bytes, &hardware_uuid) {
Ok(mut cache) => {
let backup_path = config_backup_path(&path);
if let Ok(backup_bytes) = fs::read(&backup_path)
&& let Ok(backup_cache) = decode_config_bytes(&backup_bytes, &hardware_uuid)
{
let backup_history_is_newer = backup_cache.command_history_revision
> cache.command_history_revision
|| (backup_cache.command_history_revision == 0
&& cache.command_history_revision == 0
&& cache.command_history.is_empty()
&& !backup_cache.command_history.is_empty());
if backup_history_is_newer {
cache.command_history = backup_cache.command_history;
cache.command_history_revision =
backup_cache.command_history_revision.max(1);
let encrypted_bytes = encrypt_config(&cache, &hardware_uuid)?;
persist_config_bytes(&path, &encrypted_bytes).with_context(|| {
format!(
"failed to restore command history in {} from {}",
path.display(),
backup_path.display(),
);
} else {
tracing::warn!(
"failed to parse config {} (decrypt err: {decrypt_err:#}, json err: {json_err:#}); backed up the original to {} and loaded defaults",
backup_path.display()
)
})?;
tracing::warn!(
"restored newer command history in {} from {}",
path.display(),
backup_path.display(),
);
}
}
cache
}
Err(primary_err) => {
let backup_path = config_backup_path(&path);
let backup_bytes = fs::read(&backup_path).with_context(|| {
format!("failed to read config backup {}", backup_path.display())
});
match backup_bytes.and_then(|backup_bytes| {
decode_config_bytes(&backup_bytes, &hardware_uuid)
.map(|cache| (backup_bytes, cache))
}) {
Ok((backup_bytes, cache)) => {
tracing::warn!(
"failed to load config {}: {primary_err:#}; restored {}",
path.display(),
backup_path.display(),
);
persist_config_bytes(&path, &backup_bytes).with_context(|| {
format!(
"failed to restore config {} from {}",
path.display(),
backup_path.display(),
);
}
ConfigFile::default()
backup_path.display()
)
})?;
cache
}
Err(backup_err) => {
return Err(anyhow::anyhow!(
"failed to load config {}: {primary_err:#}; backup recovery failed: {backup_err:#}",
path.display()
));
}
}
}
@@ -428,7 +661,23 @@ impl ConfigStore {
if cache.sync_device_id.is_empty() {
cache.sync_device_id = Uuid::new_v4().to_string();
}
Ok(Self { path, cache })
let mut history_changed = false;
for history in cache.command_history.values_mut() {
history_changed |= normalize_command_history_entries(history);
}
let previous_history_count = cache.command_history.len();
cache
.command_history
.retain(|_, history| !history.is_empty());
history_changed |= cache.command_history.len() != previous_history_count;
if history_changed {
cache.command_history_revision = cache.command_history_revision.saturating_add(1);
}
Ok(Self {
path,
cache,
write_lock: Arc::new(Mutex::new(())),
})
}
pub fn in_memory() -> Self {
@@ -439,6 +688,7 @@ impl ConfigStore {
Self {
path: PathBuf::new(),
cache,
write_lock: Arc::new(Mutex::new(())),
}
}
@@ -459,6 +709,92 @@ impl ConfigStore {
self.cache.sessions = sessions;
}
/// Return persisted command history for all SSH sessions, newest first per session.
pub fn all_command_history(&self) -> Vec<(String, usize, String)> {
let mut session_ids = self
.cache
.command_history
.keys()
.cloned()
.collect::<Vec<_>>();
session_ids.sort();
let mut entries = Vec::new();
for session_id in session_ids {
if let Some(history) = self.cache.command_history.get(&session_id) {
entries.extend(
history
.iter()
.enumerate()
.rev()
.map(|(index, command)| (session_id.clone(), index, command.clone())),
);
}
}
entries
}
pub fn add_command_history(&mut self, session_id: &str, command: String) -> bool {
let command = command.trim().to_string();
if command.is_empty() {
return false;
}
let history = self
.cache
.command_history
.entry(session_id.to_string())
.or_default();
let was_latest = history.last().is_some_and(|previous| previous == &command);
let previous_len = history.len();
history.retain(|previous| previous != &command);
history.push(command);
if history.len() > MAX_COMMAND_HISTORY {
let excess = history.len() - MAX_COMMAND_HISTORY;
history.drain(..excess);
}
let changed = !was_latest || history.len() != previous_len;
if changed {
self.cache.command_history_revision =
self.cache.command_history_revision.saturating_add(1);
}
changed
}
pub fn normalize_command_history(&mut self) {
let mut changed = false;
for history in self.cache.command_history.values_mut() {
changed |= normalize_command_history_entries(history);
}
let previous_history_count = self.cache.command_history.len();
self.cache
.command_history
.retain(|_, history| !history.is_empty());
changed |= self.cache.command_history.len() != previous_history_count;
if changed {
self.cache.command_history_revision =
self.cache.command_history_revision.saturating_add(1);
}
}
pub fn remove_command_history(&mut self, session_id: &str, index: usize) -> bool {
let became_empty = {
let Some(history) = self.cache.command_history.get_mut(session_id) else {
return false;
};
if index >= history.len() {
return false;
}
history.remove(index);
history.is_empty()
};
if became_empty {
self.cache.command_history.remove(session_id);
}
self.cache.command_history_revision = self.cache.command_history_revision.saturating_add(1);
true
}
pub fn sync_endpoint(&self) -> &str {
&self.cache.sync_endpoint
}
@@ -536,10 +872,6 @@ impl ConfigStore {
self.cache.sync_etag_backend = self.sync_backend().to_string();
}
pub fn tmp_dir(&self) -> Option<PathBuf> {
self.path.parent().map(|p| p.join("tmp"))
}
pub fn follow_system_theme(&self) -> bool {
self.cache.follow_system_theme
}
@@ -615,6 +947,25 @@ impl ConfigStore {
self.cache.window_bounds.as_ref()
}
pub fn remember_tabs(&self) -> bool {
self.cache.remember_tabs
}
pub fn set_remember_tabs(&mut self, remember_tabs: bool) {
self.cache.remember_tabs = remember_tabs;
if !remember_tabs {
self.cache.saved_tabs = None;
}
}
pub fn saved_tabs(&self) -> Option<&SavedTabsState> {
self.cache.saved_tabs.as_ref()
}
pub fn set_saved_tabs(&mut self, saved_tabs: Option<SavedTabsState>) {
self.cache.saved_tabs = saved_tabs;
}
pub fn workspace_panels(&self) -> Option<&Vec<f32>> {
self.cache.workspace_panels.as_ref()
}
@@ -624,6 +975,18 @@ impl ConfigStore {
self.cache.body_panels.as_ref()
}
pub fn sftp_tree_panels(&self) -> Option<&Vec<f32>> {
self.cache.sftp_tree_panels.as_ref()
}
pub fn sftp_file_columns(&self) -> Option<&Vec<f32>> {
self.cache.sftp_file_columns.as_ref()
}
pub fn sftp_file_columns_customized(&self) -> bool {
self.cache.sftp_file_columns_customized
}
pub fn transfers(&self) -> Vec<crate::terminal::Transfer> {
self.cache.transfers.clone()
}
@@ -646,6 +1009,18 @@ impl ConfigStore {
self.cache.body_panels = body_panels;
}
pub fn set_sftp_tree_panels(&mut self, panels: Option<Vec<f32>>) {
self.cache.sftp_tree_panels = panels;
}
pub fn set_sftp_file_columns(&mut self, columns: Option<Vec<f32>>) {
self.cache.sftp_file_columns = columns;
}
pub fn set_sftp_file_columns_customized(&mut self, customized: bool) {
self.cache.sftp_file_columns_customized = customized;
}
pub fn set_terminal_font_size(&mut self, terminal_font_size: f32) {
self.cache.terminal_font_size = terminal_font_size.max(10.0);
}
@@ -807,45 +1182,44 @@ impl ConfigStore {
pub fn remove(&mut self, id: &str) {
self.cache.sessions.retain(|s| s.id != id);
if self.cache.command_history.remove(id).is_some() {
self.cache.command_history_revision =
self.cache.command_history_revision.saturating_add(1);
}
}
pub fn save(&self) -> Result<()> {
if self.path.as_os_str().is_empty() {
return Ok(());
}
let _guard = self
.write_lock
.lock()
.map_err(|_| anyhow::anyhow!("configuration write lock is poisoned"))?;
let hardware_uuid = get_hardware_uuid();
let encrypted_bytes = encrypt_config(&self.cache, &hardware_uuid)?;
fs::write(&self.path, encrypted_bytes)
.with_context(|| format!("failed to write {}", self.path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
if let Ok(mut perms) = fs::metadata(&self.path).map(|m| m.permissions()) {
perms.set_mode(0o600);
let _ = fs::set_permissions(&self.path, perms);
}
}
Ok(())
write_config_bytes(&self.path, &encrypted_bytes)
}
pub fn save_merged_preferences(&self, local_config: ConfigFile) -> Result<()> {
if self.path.as_os_str().is_empty() {
return Ok(());
}
let _guard = self
.write_lock
.lock()
.map_err(|_| anyhow::anyhow!("configuration write lock is poisoned"))?;
let hardware_uuid = get_hardware_uuid();
let mut disk_config = if self.path.exists() {
if let Ok(raw_bytes) = fs::read(&self.path) {
match decrypt_config(&raw_bytes, &hardware_uuid) {
Ok(loaded) => loaded,
Err(_) => serde_json::from_slice::<ConfigFile>(&raw_bytes)
.unwrap_or_else(|_| self.cache.clone()),
}
} else {
self.cache.clone()
}
let raw_bytes = fs::read(&self.path)
.with_context(|| format!("failed to read {}", self.path.display()))?;
decode_config_bytes(&raw_bytes, &hardware_uuid).with_context(|| {
format!(
"refusing to overwrite unreadable config {}",
self.path.display()
)
})?
} else {
self.cache.clone()
};
@@ -864,29 +1238,34 @@ impl ConfigStore {
disk_config.terminal_font_family = local_config.terminal_font_family;
disk_config.title_bar_style = local_config.title_bar_style;
disk_config.cursor_style = local_config.cursor_style;
if local_config.command_history_revision >= disk_config.command_history_revision {
disk_config.command_history = local_config.command_history;
disk_config.command_history_revision = local_config.command_history_revision;
}
disk_config.remember_tabs = local_config.remember_tabs;
disk_config.saved_tabs = local_config.saved_tabs;
disk_config.window_bounds = local_config.window_bounds;
disk_config.workspace_panels = local_config.workspace_panels;
disk_config.body_panels = local_config.body_panels;
disk_config.sftp_tree_panels = local_config.sftp_tree_panels;
disk_config.sftp_file_columns = local_config.sftp_file_columns;
disk_config.sftp_file_columns_customized = local_config.sftp_file_columns_customized;
disk_config.show_hidden_files = local_config.show_hidden_files;
disk_config.lock_layout = local_config.lock_layout;
disk_config.monitoring_position = local_config.monitoring_position;
disk_config.sidebar_collapsed = local_config.sidebar_collapsed;
disk_config.sftp_panel_minimized = local_config.sftp_panel_minimized;
disk_config.key_bindings = local_config.key_bindings;
disk_config.use_proxy = local_config.use_proxy;
disk_config.read_env_proxy = local_config.read_env_proxy;
disk_config.global_proxy_type = local_config.global_proxy_type;
disk_config.global_proxy_host = local_config.global_proxy_host;
disk_config.global_proxy_port = local_config.global_proxy_port;
disk_config.global_proxy_user = local_config.global_proxy_user;
disk_config.global_proxy_password = local_config.global_proxy_password;
let encrypted_bytes = encrypt_config(&disk_config, &hardware_uuid)?;
fs::write(&self.path, encrypted_bytes)
.with_context(|| format!("failed to write {}", self.path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
if let Ok(mut perms) = fs::metadata(&self.path).map(|m| m.permissions()) {
perms.set_mode(0o600);
let _ = fs::set_permissions(&self.path, perms);
}
}
Ok(())
write_config_bytes(&self.path, &encrypted_bytes)
}
}
@@ -1099,7 +1478,7 @@ pub fn get_hardware_uuid() -> String {
#[cfg(target_os = "macos")]
{
if let Ok(output) = std::process::Command::new("ioreg")
.args(&["-rd1", "-c", "IOPlatformExpertDevice"])
.args(["-rd1", "-c", "IOPlatformExpertDevice"])
.output()
{
let stdout = String::from_utf8_lossy(&output.stdout);
@@ -1249,13 +1628,73 @@ mod tests {
assert!(decrypt_config(&encrypted, "wrong-password").is_err());
}
#[test]
fn test_remember_tabs_defaults_to_disabled_for_older_configs() {
let config: ConfigFile = serde_json::from_str("{}").unwrap();
assert!(!config.remember_tabs);
assert!(config.saved_tabs.is_none());
assert!(config.sftp_tree_panels.is_none());
assert!(config.sftp_file_columns.is_none());
assert!(!config.sftp_file_columns_customized);
}
#[test]
fn command_history_keeps_only_the_latest_duplicate() {
let mut store = ConfigStore::in_memory();
assert!(store.add_command_history("session-1", "first".to_string()));
assert!(store.add_command_history("session-1", "second".to_string()));
assert!(store.add_command_history("session-1", "first".to_string()));
assert!(!store.add_command_history("session-1", "first".to_string()));
assert_eq!(
store.cache.command_history.get("session-1"),
Some(&vec!["second".to_string(), "first".to_string()])
);
assert_eq!(store.cache.command_history_revision, 3);
}
#[test]
fn test_saved_tabs_roundtrip() {
let config = ConfigFile {
remember_tabs: true,
saved_tabs: Some(SavedTabsState {
groups: vec![SavedTabGroup {
id: "group-1".to_string(),
title: "~".to_string(),
pane_root: SavedPaneLayout::Single {
tab_id: "tab-1".to_string(),
},
tabs: vec![SavedTerminalTab::Local {
id: "tab-1".to_string(),
cwd: Some(PathBuf::from("/tmp")),
terminal_encoding: TextEncoding::Utf8,
}],
}],
active_group: Some("group-1".to_string()),
active_tab: Some("tab-1".to_string()),
}),
..Default::default()
};
let json = serde_json::to_string(&config).unwrap();
let restored: ConfigFile = serde_json::from_str(&json).unwrap();
assert!(restored.remember_tabs);
let restored_tabs = restored.saved_tabs.unwrap();
assert_eq!(restored_tabs.groups.len(), 1);
assert_eq!(restored_tabs.active_tab.as_deref(), Some("tab-1"));
}
#[test]
fn test_save_merged_preferences() {
let temp_dir = std::env::temp_dir();
let path = temp_dir.join(format!("ashell-test-config-{}.json", Uuid::new_v4()));
let temp_dir = tempfile::tempdir().unwrap();
let path = temp_dir.path().join("sessions.json");
let mut store = ConfigStore {
path: path.clone(),
cache: ConfigFile::default(),
write_lock: Arc::new(Mutex::new(())),
};
let session = Session {
@@ -1277,16 +1716,45 @@ mod tests {
proxy_password: String::new(),
protocol: "ssh".to_string(),
baud_rate: 115200,
terminal_encoding: TextEncoding::Utf8,
};
store.cache.sessions.push(session.clone());
store.save().unwrap();
let mut local_config = ConfigFile::default();
local_config.ui_font_size = 18.0;
local_config.terminal_font_size = 20.0;
local_config.show_hidden_files = true;
let mut local_config = ConfigFile {
ui_font_size: 18.0,
terminal_font_size: 20.0,
show_hidden_files: true,
sftp_file_columns_customized: true,
remember_tabs: true,
use_proxy: true,
read_env_proxy: false,
global_proxy_type: "http".to_string(),
global_proxy_host: "proxy.example.com".to_string(),
global_proxy_port: Some(8080),
global_proxy_user: "proxy-user".to_string(),
global_proxy_password: "proxy-password".to_string(),
command_history_revision: 2,
saved_tabs: Some(SavedTabsState {
groups: Vec::new(),
active_group: None,
active_tab: None,
}),
..Default::default()
};
local_config
.key_bindings
.insert("QuitApplication".to_string(), "cmd-q".to_string());
local_config.command_history.insert(
"test-session-id".to_string(),
vec!["pwd".to_string(), "ls -la".to_string()],
);
let mut stale_config = local_config.clone();
stale_config.command_history.clear();
stale_config.command_history_revision = 1;
store.save_merged_preferences(local_config).unwrap();
store.save_merged_preferences(stale_config).unwrap();
let loaded_bytes = fs::read(&path).unwrap();
let decrypted = decrypt_config(&loaded_bytes, &get_hardware_uuid()).unwrap();
@@ -1294,12 +1762,32 @@ mod tests {
assert_eq!(decrypted.ui_font_size, 18.0);
assert_eq!(decrypted.terminal_font_size, 20.0);
assert!(decrypted.show_hidden_files);
assert!(decrypted.sftp_file_columns_customized);
assert!(decrypted.remember_tabs);
assert!(decrypted.saved_tabs.is_some());
assert_eq!(
decrypted
.key_bindings
.get("QuitApplication")
.map(String::as_str),
Some("cmd-q")
);
assert!(decrypted.use_proxy);
assert!(!decrypted.read_env_proxy);
assert_eq!(decrypted.global_proxy_type, "http");
assert_eq!(decrypted.global_proxy_host, "proxy.example.com");
assert_eq!(decrypted.global_proxy_port, Some(8080));
assert_eq!(decrypted.global_proxy_user, "proxy-user");
assert_eq!(decrypted.global_proxy_password, "proxy-password");
assert_eq!(
decrypted.command_history.get("test-session-id"),
Some(&vec!["pwd".to_string(), "ls -la".to_string()])
);
assert_eq!(decrypted.command_history_revision, 2);
assert_eq!(decrypted.sessions.len(), 1);
assert_eq!(decrypted.sessions[0].name, "Test Session");
assert_eq!(decrypted.sessions[0].host, "1.2.3.4");
let _ = fs::remove_file(&path);
}
#[test]
+876 -121
View File
File diff suppressed because it is too large Load Diff
+414 -291
View File
@@ -16,10 +16,13 @@ use russh::{
client::{self, Handler},
keys::{PrivateKey, decode_secret_key, load_secret_key},
};
use russh_sftp::client::SftpSession;
use russh_sftp::{client::SftpSession, protocol::FileAttributes};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
sync::{
mpsc::{self, UnboundedReceiver, UnboundedSender},
oneshot,
},
task::JoinHandle,
};
use uuid::Uuid;
@@ -66,15 +69,22 @@ pub enum SftpCommand {
remote: String,
local_dir: String,
},
EditFile {
ReadTextFile {
remote_path: String,
reply: oneshot::Sender<std::result::Result<Vec<u8>, String>>,
},
WriteTextFile {
remote_path: String,
content: Vec<u8>,
reply: oneshot::Sender<std::result::Result<(), String>>,
},
RenamePath {
old_path: String,
new_path: String,
reply: oneshot::Sender<std::result::Result<(), String>>,
},
CreateDir(String),
DeletePaths(Vec<String>),
UploadEditedFile {
local_path: String,
remote_path: String,
},
UploadPaths {
locals: Vec<String>,
remote_dir: String,
@@ -90,6 +100,14 @@ use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
pub struct TransferStateFlag(pub Arc<AtomicU8>);
#[derive(Clone, Copy)]
struct TransferContext<'a> {
flag: &'a TransferStateFlag,
events: &'a std::sync::mpsc::Sender<BackendEvent>,
tab_id: &'a str,
id: &'a str,
}
impl TransferStateFlag {
pub fn new() -> Self {
Self(Arc::new(AtomicU8::new(0)))
@@ -184,8 +202,43 @@ impl SftpHandle {
.send(SftpCommand::UploadPaths { locals, remote_dir });
}
pub fn edit_file(&self, remote_path: String) {
let _ = self.commands.send(SftpCommand::EditFile { remote_path });
pub fn read_text_file(
&self,
remote_path: String,
) -> oneshot::Receiver<std::result::Result<Vec<u8>, String>> {
let (reply, response) = oneshot::channel();
let _ = self
.commands
.send(SftpCommand::ReadTextFile { remote_path, reply });
response
}
pub fn write_text_file(
&self,
remote_path: String,
content: Vec<u8>,
) -> oneshot::Receiver<std::result::Result<(), String>> {
let (reply, response) = oneshot::channel();
let _ = self.commands.send(SftpCommand::WriteTextFile {
remote_path,
content,
reply,
});
response
}
pub fn rename_path(
&self,
old_path: String,
new_path: String,
) -> oneshot::Receiver<std::result::Result<(), String>> {
let (reply, response) = oneshot::channel();
let _ = self.commands.send(SftpCommand::RenamePath {
old_path,
new_path,
reply,
});
response
}
pub fn close(&self) {
@@ -227,10 +280,6 @@ pub fn spawn_sftp(
tab_id: tab_id.clone(),
text: format!("sftp error: {err:#}"),
});
let _ = events.send(BackendEvent::Closed {
tab_id,
reason: format!("sftp error: {err:#}"),
});
}
});
SftpHandle {
@@ -252,17 +301,7 @@ async fn run_sftp(
});
let handle = connect_and_authenticate(&session).await?;
let channel = handle
.channel_open_session()
.await
.context("open sftp channel")?;
channel
.request_subsystem(true, "sftp")
.await
.context("request sftp subsystem")?;
let sftp = SftpSession::new(channel.into_stream())
.await
.context("sftp handshake")?;
let sftp = open_sftp_session(&handle).await?;
let home = sftp
.canonicalize(".")
@@ -310,9 +349,15 @@ async fn run_sftp(
};
if let Err(err) = emit_entries(&events, &tab_id, &sftp, &actual_path).await {
let reason = format!("list failed: {err:#}");
let _ = events.send(BackendEvent::SftpDirectoryFailed {
tab_id: tab_id.clone(),
path: actual_path,
reason: reason.clone(),
});
let _ = events.send(BackendEvent::SftpStatus {
tab_id: tab_id.clone(),
text: format!("list failed: {err:#}"),
text: reason,
});
}
}
@@ -354,33 +399,30 @@ async fn run_sftp(
let commands_tx_clone = commands_tx.clone();
tokio::spawn(async move {
let Ok(channel) = handle_clone.channel_open_session().await else {
return;
};
let Ok(_) = channel.request_subsystem(true, "sftp").await else {
return;
};
let Ok(sftp_session) = SftpSession::new(channel.into_stream()).await else {
return;
};
let result = async {
let sftp_session = open_sftp_session(&handle_clone).await?;
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: t!("downloading_file", base = base_name(&remote)).to_string(),
});
let transfer = TransferContext {
flag: &flag,
events: &events_clone,
tab_id: &tab_id_clone,
id: &id,
};
download_path_impl(
&handle_clone,
&sftp_session,
&remote,
Path::new(&local_dir),
transfer,
)
.await
}
.await;
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: t!("downloading_file", base = base_name(&remote)).to_string(),
});
match download_path_impl(
&handle_clone,
&sftp_session,
&remote,
Path::new(&local_dir),
flag,
&events_clone,
&tab_id_clone,
&id,
)
.await
{
match result {
Ok(summary) => {
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone,
@@ -467,32 +509,26 @@ async fn run_sftp(
let commands_tx_clone = commands_tx.clone();
tokio::spawn(async move {
let Ok(channel) = handle_clone.channel_open_session().await else {
return;
};
let Ok(_) = channel.request_subsystem(true, "sftp").await else {
return;
};
let Ok(sftp_session) = SftpSession::new(channel.into_stream()).await else {
return;
};
let result = async {
let sftp_session = open_sftp_session(&handle_clone).await?;
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: t!("uploading").to_string(),
});
upload_paths_impl(
&sftp_session,
&locals,
&remote_dir,
flag,
&events_clone,
&tab_id_clone,
&id,
)
.await
}
.await;
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: t!("uploading").to_string(),
});
match upload_paths_impl(
&sftp_session,
&locals,
&remote_dir,
flag,
&events_clone,
&tab_id_clone,
&id,
)
.await
{
match result {
Ok(summary) => {
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
@@ -530,150 +566,41 @@ async fn run_sftp(
let _ = commands_tx_clone.send(SftpCommand::TransferFinished(id));
});
}
SftpCommand::EditFile { remote_path } => {
let id = uuid::Uuid::new_v4().to_string();
let config = crate::session::config::ConfigStore::load().unwrap();
let tmp_dir = config.tmp_dir().unwrap_or_else(|| PathBuf::from("/tmp"));
let base = base_name(&remote_path);
let local_path = tmp_dir.join(format!("{}-{}", id, base));
let handle_clone = handle.clone();
let commands_tx_clone = commands_tx.clone();
let events_clone = events.clone();
let tab_id_clone = tab_id.clone();
tokio::spawn(async move {
let flag = TransferStateFlag::new();
let Ok(channel) = handle_clone.channel_open_session().await else {
return;
};
let Ok(_) = channel.request_subsystem(true, "sftp").await else {
return;
};
let Ok(sftp_session) = SftpSession::new(channel.into_stream()).await else {
return;
};
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: t!("downloading_file", base = base).to_string(),
});
if let Err(err) = download_file_impl(
&sftp_session,
&remote_path,
&local_path,
&flag,
&events_clone,
&tab_id_clone,
"edit-download",
)
SftpCommand::ReadTextFile { remote_path, reply } => {
let result = read_text_file_impl(&sftp, &remote_path)
.await
{
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: format!("Edit download failed: {err:#}"),
});
return;
}
if let Err(err) = open::that(&local_path) {
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: format!("Failed to open editor: {err:#}"),
});
return;
}
use notify::Watcher;
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let mut watcher = match notify::recommended_watcher(
move |res: notify::Result<notify::Event>| {
if let Ok(event) = res {
if event.kind.is_modify() {
let _ = tx.send(());
}
}
},
) {
Ok(w) => w,
Err(_) => return,
};
if let Err(_) = watcher.watch(&local_path, notify::RecursiveMode::NonRecursive)
{
return;
}
while let Some(_) = rx.recv().await {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
while let Ok(_) = rx.try_recv() {} // drain pending
if commands_tx_clone
.send(SftpCommand::UploadEditedFile {
local_path: local_path.to_string_lossy().to_string(),
remote_path: remote_path.clone(),
})
.is_err()
{
break;
}
}
});
.map_err(|err| format!("{err:#}"));
let _ = reply.send(result);
}
SftpCommand::UploadEditedFile {
local_path,
SftpCommand::WriteTextFile {
remote_path,
content,
reply,
} => {
let handle_clone = handle.clone();
let events_clone = events.clone();
let tab_id_clone = tab_id.clone();
tokio::spawn(async move {
let flag = TransferStateFlag::new();
let Ok(channel) = handle_clone.channel_open_session().await else {
return;
};
let Ok(_) = channel.request_subsystem(true, "sftp").await else {
return;
};
let Ok(sftp_session) = SftpSession::new(channel.into_stream()).await else {
return;
};
let transferred = Arc::new(AtomicU64::new(0));
match upload_file_impl(
&sftp_session,
Path::new(&local_path),
&remote_path,
&flag,
&events_clone,
&tab_id_clone,
"edit-upload",
transferred,
None,
)
let result = write_text_file_impl(&sftp, &remote_path, &content)
.await
{
Ok(_) => {
let now = chrono::Local::now().format("%H:%M:%S");
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: format!(
"{} ({})",
t!("auto_saved_and_uploaded", base = base_name(&remote_path)),
now
),
});
}
Err(err) => {
let _ = events_clone.send(BackendEvent::SftpStatus {
tab_id: tab_id_clone.clone(),
text: format!("Auto-upload failed: {err:#}"),
});
}
.map_err(|err| format!("{err:#}"));
if result.is_ok() {
if let Some(parent) = parent_dir(&remote_path) {
let _ = commands_tx.send(SftpCommand::ListDir(parent));
}
});
}
let _ = reply.send(result);
}
SftpCommand::RenamePath {
old_path,
new_path,
reply,
} => {
let result = rename_path_impl(&sftp, &old_path, &new_path)
.await
.map_err(|err| format!("{err:#}"));
if result.is_ok() {
if let Some(parent) = parent_dir(&old_path) {
let _ = commands_tx.send(SftpCommand::ListDir(parent));
}
}
let _ = reply.send(result);
}
SftpCommand::CreateDir(path) => {
let actual_path = if path == "~" {
@@ -767,6 +694,22 @@ async fn run_sftp(
Ok(())
}
async fn open_sftp_session(
handle: &russh::client::Handle<SftpClientHandler>,
) -> Result<SftpSession> {
let channel = handle
.channel_open_session()
.await
.context("open sftp channel")?;
channel
.request_subsystem(true, "sftp")
.await
.context("request sftp subsystem")?;
SftpSession::new(channel.into_stream())
.await
.context("sftp handshake")
}
use std::future::Future;
use std::pin::Pin;
@@ -851,7 +794,7 @@ async fn connect_and_authenticate(
.context("password authentication failed")?,
AuthMethod::Key => {
let has_explicit_key = session_has_explicit_key(session);
let success = if has_explicit_key {
if has_explicit_key {
let keypair = load_session_private_key(session)?;
let keys = private_keys_with_algs(keypair).context("invalid private key")?;
let mut success = false;
@@ -896,8 +839,7 @@ async fn connect_and_authenticate(
));
}
success
};
success
}
}
AuthMethod::Config => {
// For Config auth, try the identity file from config entry, or default keys
@@ -1017,7 +959,7 @@ fn expand_key_path(value: &str) -> Option<PathBuf> {
Some(Path::new(value).to_path_buf())
}
fn base_name(path: &str) -> String {
pub(crate) fn base_name(path: &str) -> String {
let sep = |c: char| c == '/' || c == '\\';
path.trim_end_matches(sep)
.rsplit(sep)
@@ -1026,6 +968,50 @@ fn base_name(path: &str) -> String {
.to_string()
}
pub(crate) fn editor_language(path: &str) -> &'static str {
let name = base_name(path).to_lowercase();
if name == "dockerfile" {
return "bash";
}
if name == "makefile" {
return "make";
}
match Path::new(&name)
.extension()
.and_then(|extension| extension.to_str())
.unwrap_or_default()
{
"bash" | "zsh" | "sh" => "bash",
"c" | "h" => "c",
"cc" | "cpp" | "cxx" | "hpp" => "cpp",
"cs" => "csharp",
"css" | "scss" => "css",
"go" => "go",
"html" | "htm" => "html",
"java" => "java",
"js" | "mjs" | "cjs" => "javascript",
"json" | "jsonc" => "json",
"kt" | "kts" => "kotlin",
"lua" => "lua",
"md" | "markdown" => "markdown",
"php" => "php",
"proto" => "proto",
"py" => "python",
"rb" => "ruby",
"rs" => "rust",
"sql" => "sql",
"svelte" => "svelte",
"swift" => "swift",
"toml" => "toml",
"ts" => "typescript",
"tsx" => "tsx",
"yaml" | "yml" => "yaml",
"zig" => "zig",
_ => "text",
}
}
pub(crate) fn parent_dir(path: &str) -> Option<String> {
if path == "/" || path.is_empty() {
return None;
@@ -1050,6 +1036,53 @@ pub(crate) fn join_remote(parent: &str, child: &str) -> String {
}
}
pub(crate) fn normalize_remote_path(input: &str, current: &str, home: &str) -> String {
let input = input.trim();
let expanded = if input.is_empty() {
current.to_string()
} else if input == "~" {
home.to_string()
} else if let Some(rest) = input.strip_prefix("~/") {
join_remote(home, rest)
} else if input.starts_with('/') {
input.to_string()
} else {
join_remote(current, input)
};
let mut components = Vec::new();
for component in expanded.split('/') {
match component {
"" | "." => {}
".." => {
components.pop();
}
value => components.push(value),
}
}
if components.is_empty() {
"/".to_string()
} else {
format!("/{}", components.join("/"))
}
}
pub(crate) fn remote_path_ancestors(path: &str) -> Vec<String> {
let normalized = normalize_remote_path(path, "/", "/");
let mut ancestors = vec!["/".to_string()];
let mut current = String::new();
for component in normalized.trim_start_matches('/').split('/') {
if component.is_empty() {
continue;
}
current.push('/');
current.push_str(component);
ancestors.push(current.clone());
}
ancestors
}
#[allow(dead_code)]
fn strip_archive_suffix(name: &str) -> &str {
for suffix in [".tar.gz", ".tgz", ".zip", ".tar"] {
@@ -1119,6 +1152,97 @@ async fn list_dir_impl(sftp: &SftpSession, path: &str) -> Result<Vec<RemoteEntry
Ok(entries)
}
pub(crate) const MAX_INLINE_EDIT_BYTES: usize = 2 * 1024 * 1024;
async fn read_text_file_impl(sftp: &SftpSession, path: &str) -> Result<Vec<u8>> {
let metadata = sftp
.metadata(path)
.await
.with_context(|| format!("read metadata for {path}"))?;
if metadata.size.unwrap_or(0) > MAX_INLINE_EDIT_BYTES as u64 {
return Err(anyhow!("file exceeds the 2 MB in-app editing limit"));
}
let remote_file = sftp
.open(path)
.await
.with_context(|| format!("open remote {path}"))?;
let mut limited = remote_file.take((MAX_INLINE_EDIT_BYTES + 1) as u64);
let mut content = Vec::new();
limited
.read_to_end(&mut content)
.await
.with_context(|| format!("read remote {path}"))?;
if content.len() > MAX_INLINE_EDIT_BYTES {
return Err(anyhow!("file exceeds the 2 MB in-app editing limit"));
}
Ok(content)
}
async fn write_text_file_impl(sftp: &SftpSession, path: &str, content: &[u8]) -> Result<()> {
if content.len() > MAX_INLINE_EDIT_BYTES {
return Err(anyhow!("file exceeds the 2 MB in-app editing limit"));
}
let permissions = sftp
.metadata(path)
.await
.ok()
.and_then(|meta| meta.permissions);
let temporary_path = format!("{path}.ashell-{}.tmp", Uuid::new_v4());
let write_result = async {
let mut remote_file = sftp
.create(temporary_path.as_str())
.await
.with_context(|| format!("open remote temporary file {temporary_path}"))?;
remote_file
.write_all(content)
.await
.with_context(|| format!("write remote temporary file {temporary_path}"))?;
remote_file
.flush()
.await
.with_context(|| format!("flush remote temporary file {temporary_path}"))?;
drop(remote_file);
if let Some(permissions) = permissions {
sftp.set_metadata(
temporary_path.as_str(),
FileAttributes {
permissions: Some(permissions),
..FileAttributes::default()
},
)
.await
.with_context(|| format!("preserve permissions for {path}"))?;
}
sftp.rename(temporary_path.as_str(), path)
.await
.with_context(|| format!("replace remote {path}"))
}
.await;
if let Err(err) = write_result {
let _ = sftp.remove_file(temporary_path.as_str()).await;
return Err(err);
}
Ok(())
}
async fn rename_path_impl(sftp: &SftpSession, old_path: &str, new_path: &str) -> Result<()> {
if old_path == new_path {
return Ok(());
}
if sftp.metadata(new_path).await.is_ok() {
return Err(anyhow!("target already exists: {new_path}"));
}
sftp.rename(old_path, new_path)
.await
.with_context(|| format!("rename {old_path} to {new_path}"))
}
async fn preview_impl(sftp: &SftpSession, path: &str) -> Result<PreviewData> {
let metadata = sftp
.metadata(path)
@@ -1183,17 +1307,14 @@ async fn download_path_impl(
sftp: &SftpSession,
remote: &str,
local_dir: &Path,
flag: TransferStateFlag,
events: &std::sync::mpsc::Sender<BackendEvent>,
tab_id: &str,
id: &str,
transfer: TransferContext<'_>,
) -> Result<String> {
tokio::fs::create_dir_all(local_dir)
.await
.with_context(|| format!("create {}", local_dir.display()))?;
// Check for cancellation after initial setup
let state = flag.0.load(Ordering::SeqCst);
let state = transfer.flag.0.load(Ordering::SeqCst);
if state == 2 {
return Err(anyhow::anyhow!("transfer cancelled"));
}
@@ -1213,22 +1334,14 @@ async fn download_path_impl(
base_name(remote),
Uuid::new_v4()
));
let extracted_to = download_remote_directory_archive(
handle,
sftp,
remote,
&local_archive,
&flag,
events,
tab_id,
id,
)
.await?;
let extracted_to =
download_remote_directory_archive(handle, sftp, remote, &local_archive, transfer)
.await?;
return Ok(t!("downloaded_folder", path = extracted_to.display()).to_string());
}
let local_path = local_dir.join(base_name(remote));
download_file_impl(sftp, remote, &local_path, &flag, events, tab_id, id).await?;
download_file_impl(sftp, remote, &local_path, transfer).await?;
Ok(t!("downloaded_file", path = local_path.display()).to_string())
}
@@ -1237,10 +1350,7 @@ async fn download_dir_recursive(
sftp: &SftpSession,
remote_dir: &str,
local_dir: &Path,
flag: &TransferStateFlag,
events: &std::sync::mpsc::Sender<BackendEvent>,
tab_id: &str,
id: &str,
transfer: TransferContext<'_>,
) -> Result<()> {
tokio::fs::create_dir_all(local_dir)
.await
@@ -1253,23 +1363,11 @@ async fn download_dir_recursive(
sftp,
&entry.full_path,
&local_path,
flag,
events,
tab_id,
id,
transfer,
))
.await?;
} else {
download_file_impl(
sftp,
&entry.full_path,
&local_path,
flag,
events,
tab_id,
id,
)
.await?;
download_file_impl(sftp, &entry.full_path, &local_path, transfer).await?;
let _ = maybe_extract_archive(&local_path).await;
}
}
@@ -1281,10 +1379,7 @@ async fn download_remote_directory_archive(
sftp: &SftpSession,
remote_dir: &str,
local_archive: &Path,
flag: &TransferStateFlag,
events: &std::sync::mpsc::Sender<BackendEvent>,
tab_id: &str,
id: &str,
transfer: TransferContext<'_>,
) -> Result<PathBuf> {
let remote_archive = format!(
"/tmp/ashell-{}-{}.tar.gz",
@@ -1293,7 +1388,7 @@ async fn download_remote_directory_archive(
);
// Check for cancellation before creating remote archive
let state = flag.0.load(Ordering::SeqCst);
let state = transfer.flag.0.load(Ordering::SeqCst);
if state == 2 {
return Err(anyhow::anyhow!("transfer cancelled"));
}
@@ -1306,16 +1401,7 @@ async fn download_remote_directory_archive(
.join(base_name(remote_dir));
let archive_download = async {
download_file_impl(
sftp,
&remote_archive,
local_archive,
flag,
events,
tab_id,
id,
)
.await?;
download_file_impl(sftp, &remote_archive, local_archive, transfer).await?;
extract_archive_to(
local_archive,
local_archive.parent().unwrap_or_else(|| Path::new(".")),
@@ -1342,10 +1428,7 @@ async fn download_file_impl(
sftp: &SftpSession,
remote: &str,
local: &Path,
flag: &TransferStateFlag,
events: &std::sync::mpsc::Sender<BackendEvent>,
tab_id: &str,
id: &str,
transfer: TransferContext<'_>,
) -> Result<()> {
let mut remote_file = sftp
.open(remote)
@@ -1360,7 +1443,15 @@ async fn download_file_impl(
let mut buffer = vec![0u8; 128 * 1024];
loop {
flag.yield_if_paused(events, tab_id, id, transferred, total)
transfer
.flag
.yield_if_paused(
transfer.events,
transfer.tab_id,
transfer.id,
transferred,
total,
)
.await?;
let read = remote_file
.read(&mut buffer)
@@ -1375,9 +1466,9 @@ async fn download_file_impl(
.with_context(|| format!("write {}", local.display()))?;
transferred += read as u64;
let _ = events.send(BackendEvent::TransferProgress {
tab_id: tab_id.to_string(),
id: id.to_string(),
let _ = transfer.events.send(BackendEvent::TransferProgress {
tab_id: transfer.tab_id.to_string(),
id: transfer.id.to_string(),
transferred,
total,
state: crate::terminal::TransferState::Running,
@@ -1385,9 +1476,9 @@ async fn download_file_impl(
}
local_file.flush().await.context("flush local file")?;
let _ = events.send(BackendEvent::TransferProgress {
tab_id: tab_id.to_string(),
id: id.to_string(),
let _ = transfer.events.send(BackendEvent::TransferProgress {
tab_id: transfer.tab_id.to_string(),
id: transfer.id.to_string(),
transferred,
total,
state: crate::terminal::TransferState::Completed,
@@ -1490,14 +1581,17 @@ async fn upload_paths_impl(
let transferred_clone = Arc::clone(&transferred);
futures.push(async move {
let transfer = TransferContext {
flag: &flag_clone,
events: &events_clone,
tab_id: &tab_id_clone,
id: &id_clone,
};
upload_file_impl(
sftp,
&local_path,
&remote_path,
&flag_clone,
&events_clone,
&tab_id_clone,
&id_clone,
transfer,
transferred_clone,
Some(total_bytes),
)
@@ -1542,10 +1636,7 @@ async fn upload_file_impl(
sftp: &SftpSession,
local_file: &Path,
remote_path: &str,
flag: &TransferStateFlag,
events: &std::sync::mpsc::Sender<BackendEvent>,
tab_id: &str,
id: &str,
transfer: TransferContext<'_>,
transferred: Arc<AtomicU64>,
total: Option<u64>,
) -> Result<()> {
@@ -1560,7 +1651,10 @@ async fn upload_file_impl(
let mut buffer = vec![0u8; 128 * 1024];
loop {
let cur = transferred.load(Ordering::Relaxed);
flag.yield_if_paused(events, tab_id, id, cur, total).await?;
transfer
.flag
.yield_if_paused(transfer.events, transfer.tab_id, transfer.id, cur, total)
.await?;
let read = local.read(&mut buffer).await.context("read local file")?;
if read == 0 {
break;
@@ -1571,9 +1665,9 @@ async fn upload_file_impl(
.with_context(|| format!("write remote {remote_path}"))?;
let new_cur = transferred.fetch_add(read as u64, Ordering::Relaxed) + read as u64;
let _ = events.send(BackendEvent::TransferProgress {
tab_id: tab_id.to_string(),
id: id.to_string(),
let _ = transfer.events.send(BackendEvent::TransferProgress {
tab_id: transfer.tab_id.to_string(),
id: transfer.id.to_string(),
transferred: new_cur,
total,
state: crate::terminal::TransferState::Running,
@@ -1845,3 +1939,32 @@ impl Handler for SftpClientHandler {
Ok(true)
}
}
#[cfg(test)]
mod path_tests {
use super::{normalize_remote_path, remote_path_ancestors};
#[test]
fn normalizes_remote_paths_without_platform_separators() {
assert_eq!(
normalize_remote_path("~", "/tmp", "/home/demo"),
"/home/demo"
);
assert_eq!(
normalize_remote_path("../logs", "/srv/app/current", "/home/demo"),
"/srv/app/logs"
);
assert_eq!(
normalize_remote_path("/var//log/", "/", "/home/demo"),
"/var/log"
);
}
#[test]
fn builds_remote_path_ancestors_from_root() {
assert_eq!(
remote_path_ancestors("/home/demo/projects"),
vec!["/", "/home", "/home/demo", "/home/demo/projects"]
);
}
}
+123 -33
View File
@@ -6,26 +6,6 @@ use crate::{
terminal,
};
pub(crate) fn is_editable_text_file(filename: &str) -> bool {
let lower = filename.to_lowercase();
let ext = std::path::Path::new(&lower)
.extension()
.and_then(|s| s.to_str())
.unwrap_or("");
let known_exts = [
"txt", "conf", "json", "yaml", "yml", "xml", "ini", "sh", "py", "rs", "js", "ts", "html",
"css", "md", "toml", "csv", "log", "cfg",
];
if known_exts.contains(&ext) {
return true;
}
let known_names = ["dockerfile", "makefile", ".gitignore", ".env"];
if known_names.contains(&lower.as_str()) {
return true;
}
false
}
impl Ashell {
pub(crate) fn active_sftp(&self) -> Option<&terminal::SftpUiState> {
self.active_group
@@ -49,13 +29,107 @@ impl Ashell {
}
pub(crate) fn navigate_sftp(&mut self, path: String, cx: &mut Context<Self>) {
if let Some(handle) = self.active_sftp_handle() {
tracing::info!("[sftp] navigating to directory: '{}'", path);
handle.list_dir(path.clone());
if let Some(sftp) = self.active_sftp_mut() {
sftp.current_path = path;
self.pending_sftp_path_sync = Some(sftp.current_path.clone());
let Some((current_path, home_dir)) = self
.active_sftp()
.map(|sftp| (sftp.current_path.clone(), sftp.home_dir.clone()))
else {
return;
};
let path = crate::sftp::normalize_remote_path(&path, &current_path, &home_dir);
let Some(handle) = self.active_sftp_handle().cloned() else {
return;
};
tracing::info!("[sftp] navigating to directory: '{}'", path);
let ancestors = crate::sftp::remote_path_ancestors(&path);
let mut paths_to_load = Vec::new();
if let Some(sftp) = self.active_sftp_mut() {
sftp.current_path = path.clone();
sftp.selected_path = None;
sftp.preview = None;
sftp.selected_entries.clear();
sftp.expand_to(&path);
sftp.begin_directory_load(&path);
for ancestor in ancestors {
if ancestor != path
&& !sftp.directory_cache.contains_key(&ancestor)
&& !sftp.loading_directories.contains(&ancestor)
{
sftp.begin_directory_load(&ancestor);
paths_to_load.push(ancestor);
}
}
}
self.pending_sftp_path_sync = Some(path.clone());
for ancestor in paths_to_load {
handle.list_dir(ancestor);
}
handle.list_dir(path);
cx.notify();
}
pub(crate) fn toggle_sftp_tree_directory(&mut self, path: String, cx: &mut Context<Self>) {
let Some(handle) = self.active_sftp_handle().cloned() else {
return;
};
let mut should_load = false;
if let Some(sftp) = self.active_sftp_mut() {
let has_error = sftp.directory_errors.contains_key(&path);
let has_cache = sftp.directory_cache.contains_key(&path);
if sftp.expanded_directories.contains(&path) && !has_error && has_cache {
sftp.expanded_directories.remove(&path);
cx.notify();
return;
}
sftp.expanded_directories.insert(path.clone());
should_load = (has_error || !has_cache) && !sftp.loading_directories.contains(&path);
if should_load {
sftp.begin_directory_load(&path);
}
}
if should_load {
handle.list_dir(path);
}
cx.notify();
}
pub(crate) fn reveal_current_sftp_directory(&mut self, cx: &mut Context<Self>) {
let Some(handle) = self.active_sftp_handle().cloned() else {
return;
};
let Some(current_path) = self.active_sftp().map(|sftp| sftp.current_path.clone()) else {
return;
};
let ancestors = crate::sftp::remote_path_ancestors(&current_path);
let mut paths_to_load = Vec::new();
if let Some(sftp) = self.active_sftp_mut() {
sftp.expand_to(&current_path);
for path in ancestors {
if !sftp.directory_cache.contains_key(&path)
&& !sftp.loading_directories.contains(&path)
{
sftp.begin_directory_load(&path);
paths_to_load.push(path);
}
}
}
for path in paths_to_load {
handle.list_dir(path);
}
if let Some(index) = self.active_sftp().and_then(|sftp| {
sftp.tree_rows(self.show_hidden_files)
.iter()
.position(|row| row.path == current_path)
}) {
self.remote_tree_scroll_handle
.scroll_to_item(index, gpui::ScrollStrategy::Center);
}
cx.notify();
}
pub(crate) fn collapse_sftp_tree(&mut self, cx: &mut Context<Self>) {
if let Some(sftp) = self.active_sftp_mut() {
sftp.collapse_all();
cx.notify();
}
}
@@ -144,15 +218,27 @@ impl Ashell {
cx.notify();
}
pub(crate) fn trigger_sftp_context_edit(&mut self, cx: &mut Context<Self>) {
pub(crate) fn trigger_sftp_context_edit(
&mut self,
window: &mut Window,
cx: &mut Context<Self>,
) {
let Some(menu) = self.sftp_context_menu.take() else {
return;
};
if let Some(handle) = self.active_sftp_handle() {
tracing::info!("[sftp] triggering edit for file: '{}'", menu.remote_path);
handle.edit_file(menu.remote_path);
}
cx.notify();
tracing::info!("[sftp] opening in-app editor for: '{}'", menu.remote_path);
self.show_sftp_editor_dialog(menu.remote_path, window, cx);
}
pub(crate) fn trigger_sftp_context_rename(
&mut self,
window: &mut Window,
cx: &mut Context<Self>,
) {
let Some(menu) = self.sftp_context_menu.take() else {
return;
};
self.show_sftp_rename_dialog(menu.remote_path, window, cx);
}
pub(crate) fn download_sftp_entry(
@@ -307,7 +393,11 @@ impl Ashell {
pub(crate) fn toggle_all_sftp_entries(&mut self, checked: bool, cx: &mut Context<Self>) {
if let Some(sftp) = self.active_sftp_mut() {
if checked {
let paths: Vec<String> = sftp.entries.iter().map(|e| e.full_path.clone()).collect();
let paths: Vec<String> = sftp
.current_entries()
.iter()
.map(|entry| entry.full_path.clone())
.collect();
for path in paths {
sftp.selected_entries.insert(path);
}
+159 -4
View File
@@ -22,6 +22,25 @@ pub struct DiskSample {
pub total_bytes: u64,
}
#[derive(Debug, Clone, Default)]
pub struct RemoteProcess {
pub pid: u32,
pub user: String,
pub cpu_percent: f32,
pub memory_bytes: u64,
pub command: String,
}
#[derive(Debug, Clone, Default)]
pub struct RemotePort {
pub protocol: String,
pub address: String,
pub port: u16,
pub state: String,
pub pid: Option<u32>,
pub process: String,
}
#[derive(Debug, Clone, Default)]
pub struct SystemSnapshot {
pub cpu_percent: f32,
@@ -52,8 +71,8 @@ impl SystemSampler {
sys.refresh_all();
let nets = Networks::new_with_refreshed_list();
let disks = Disks::new_with_refreshed_list();
let last_rx_total = nets.iter().map(|(_, d)| d.total_received()).sum();
let last_tx_total = nets.iter().map(|(_, d)| d.total_transmitted()).sum();
let last_rx_total = nets.values().map(|d| d.total_received()).sum();
let last_tx_total = nets.values().map(|d| d.total_transmitted()).sum();
Self {
sys,
@@ -81,8 +100,8 @@ impl SystemSampler {
let swap_total = self.sys.total_swap();
let swap_used = self.sys.used_swap();
let rx_total: u64 = self.nets.iter().map(|(_, d)| d.total_received()).sum();
let tx_total: u64 = self.nets.iter().map(|(_, d)| d.total_transmitted()).sum();
let rx_total: u64 = self.nets.values().map(|d| d.total_received()).sum();
let tx_total: u64 = self.nets.values().map(|d| d.total_transmitted()).sum();
let now = Instant::now();
let elapsed = now
.duration_since(self.last_instant)
@@ -228,8 +247,144 @@ pub fn remote_snapshot_from_kv(raw: &str) -> Result<SystemSnapshot> {
})
}
pub fn remote_processes_from_ps(raw: &str) -> Vec<RemoteProcess> {
raw.lines()
.filter_map(|line| {
if let Some(record) = line.strip_prefix("PROCESS\t") {
let mut fields = record.splitn(5, '\t');
let pid = fields.next()?.parse::<u32>().ok()?;
let user = fields.next()?.to_string();
let cpu_percent = fields.next()?.parse::<f32>().ok()?.max(0.0);
let memory_bytes = fields.next()?.parse::<u64>().ok()?;
let command = fields.next()?.trim().to_string();
return (!command.is_empty()).then_some(RemoteProcess {
pid,
user,
cpu_percent,
memory_bytes,
command,
});
}
let mut fields = line.split_whitespace();
let pid = fields.next()?.parse::<u32>().ok()?;
let user = fields.next()?.to_string();
let cpu_percent = fields.next()?.parse::<f32>().ok()?.max(0.0);
let memory_bytes = fields.next()?.parse::<u64>().ok()?.saturating_mul(1024);
let command = fields.collect::<Vec<_>>().join(" ");
if command.is_empty() {
return None;
}
Some(RemoteProcess {
pid,
user,
cpu_percent,
memory_bytes,
command,
})
})
.collect()
}
pub fn remote_ports_from_probe(raw: &str) -> Vec<RemotePort> {
raw.lines()
.filter_map(|line| {
let record = line.strip_prefix("PORT\t")?;
let mut fields = record.splitn(6, '\t');
let protocol = fields.next()?.trim().to_string();
let address = fields.next()?.trim().to_string();
let port = fields.next()?.parse::<u16>().ok()?;
let state = fields.next()?.trim().to_string();
let pid = match fields.next()?.trim() {
"" | "-" | "0" => None,
value => value.parse::<u32>().ok(),
};
let process = fields.next()?.trim().to_string();
if protocol.is_empty() || address.is_empty() || state.is_empty() {
return None;
}
Some(RemotePort {
protocol,
address,
port,
state,
pid,
process: if process.is_empty() {
"-".to_string()
} else {
process
},
})
})
.collect()
}
fn parse_u64(kv: &BTreeMap<String, String>, key: &str) -> u64 {
kv.get(key)
.and_then(|value| value.parse::<u64>().ok())
.unwrap_or_default()
}
#[cfg(test)]
mod tests {
use super::{remote_ports_from_probe, remote_processes_from_ps};
#[test]
fn parses_remote_process_rows_and_rss_bytes() {
let processes = remote_processes_from_ps(
" 42 alice 12.5 2048 worker --queue main\n 7 root 0.0 512 sshd\n",
);
assert_eq!(processes.len(), 2);
assert_eq!(processes[0].pid, 42);
assert_eq!(processes[0].user, "alice");
assert_eq!(processes[0].cpu_percent, 12.5);
assert_eq!(processes[0].memory_bytes, 2 * 1024 * 1024);
assert_eq!(processes[0].command, "worker --queue main");
}
#[test]
fn parses_structured_current_process_rows() {
let processes =
remote_processes_from_ps("PROCESS\t42\talice\t37.25\t2097152\tworker --queue main\n");
assert_eq!(processes.len(), 1);
assert_eq!(processes[0].pid, 42);
assert_eq!(processes[0].user, "alice");
assert_eq!(processes[0].cpu_percent, 37.25);
assert_eq!(processes[0].memory_bytes, 2 * 1024 * 1024);
assert_eq!(processes[0].command, "worker --queue main");
}
#[test]
fn skips_malformed_remote_process_rows() {
let processes = remote_processes_from_ps("PID USER CPU RSS COMMAND\n9 root bad 64 init\n");
assert!(processes.is_empty());
}
#[test]
fn parses_structured_remote_port_rows() {
let ports = remote_ports_from_probe(
"PORT\ttcp\t0.0.0.0\t22\tLISTEN\t123\tsshd\nPORT\tudp\t127.0.0.1\t5353\tUNCONN\t-\t-\n",
);
assert_eq!(ports.len(), 2);
assert_eq!(ports[0].protocol, "tcp");
assert_eq!(ports[0].address, "0.0.0.0");
assert_eq!(ports[0].port, 22);
assert_eq!(ports[0].state, "LISTEN");
assert_eq!(ports[0].pid, Some(123));
assert_eq!(ports[0].process, "sshd");
assert_eq!(ports[1].pid, None);
}
#[test]
fn skips_malformed_remote_port_rows() {
let ports = remote_ports_from_probe(
"PORT\ttcp\t0.0.0.0\tbad\tLISTEN\t1\tsshd\nPORT\ttcp\t\t80\tLISTEN\t1\tnginx\n",
);
assert!(ports.is_empty());
}
}
+27 -13
View File
@@ -156,6 +156,19 @@ pub struct TerminalElement {
search_highlights: Option<std::collections::HashMap<(i32, i32), Hsla>>,
}
pub(crate) struct TerminalElementConfig {
pub(crate) view: Entity<Ashell>,
pub(crate) focus_handle: FocusHandle,
pub(crate) snapshot: RenderSnapshot,
pub(crate) marked_text: Option<String>,
pub(crate) font_family: SharedString,
pub(crate) font_size: Pixels,
pub(crate) line_height: Pixels,
pub(crate) cell_width: Pixels,
pub(crate) tab_id: String,
pub(crate) search_highlights: Option<std::collections::HashMap<(i32, i32), Hsla>>,
}
pub struct PrepaintState {
bounds: Bounds<Pixels>,
metrics: TerminalMetrics,
@@ -284,18 +297,19 @@ impl InputHandler for TerminalInputHandler {
}
impl TerminalElement {
pub fn new(
view: Entity<Ashell>,
focus_handle: FocusHandle,
snapshot: RenderSnapshot,
marked_text: Option<String>,
font_family: SharedString,
font_size: Pixels,
line_height: Pixels,
cell_width: Pixels,
tab_id: String,
search_highlights: Option<std::collections::HashMap<(i32, i32), Hsla>>,
) -> Self {
pub(crate) fn new(config: TerminalElementConfig) -> Self {
let TerminalElementConfig {
view,
focus_handle,
snapshot,
marked_text,
font_family,
font_size,
line_height,
cell_width,
tab_id,
search_highlights,
} = config;
Self {
view,
focus_handle,
@@ -606,7 +620,7 @@ impl Element for TerminalElement {
// This is 100% accurate because it is recorded during layout prepaint.
let view = self.view.clone();
let tab_id = self.tab_id.clone();
let _ = view.update(cx, |this, cx| {
view.update(cx, |this, cx| {
let old_bounds = this.terminal_bounds.insert(tab_id.clone(), bounds);
// Sync PTY size unconditionally on every prepaint layout pass to ensure
+6 -6
View File
@@ -877,10 +877,10 @@ fn find_urls(text: &str) -> Vec<usize> {
while let Some(pos) = text[start..].find("http") {
let abs = start + pos;
let remaining = &text[abs..];
if remaining.starts_with("https://") || remaining.starts_with("http://") {
if abs == 0 || is_boundary(text.as_bytes()[abs - 1] as char) {
positions.push(abs);
}
if (remaining.starts_with("https://") || remaining.starts_with("http://"))
&& (abs == 0 || is_boundary(text.as_bytes()[abs - 1] as char))
{
positions.push(abs);
}
start = abs + 4;
}
@@ -967,8 +967,8 @@ pub fn build_logical_lines<'a>(cells: &'a [RenderCell], rows: usize) -> Vec<Logi
}
let wraps_from_prev = row_idx > 0 && {
current_line.as_ref().map_or(false, |line| {
line.row_cells.last().map_or(false, |rc| {
current_line.as_ref().is_some_and(|line| {
line.row_cells.last().is_some_and(|rc| {
rc.cell
.flags
.contains(alacritty_terminal::term::cell::Flags::WRAPLINE)
+230 -27
View File
@@ -13,7 +13,7 @@ use crate::{
};
thread_local! {
static LAST_DRAG_SCROLL: std::cell::Cell<Option<std::time::Instant>> = std::cell::Cell::new(None);
static LAST_DRAG_SCROLL: std::cell::Cell<Option<std::time::Instant>> = const { std::cell::Cell::new(None) };
}
impl Ashell {
@@ -42,10 +42,10 @@ impl Ashell {
&& !event.keystroke.modifiers.platform
{
match event.keystroke.key.to_ascii_lowercase().as_str() {
"h" => self.focus_adjacent_pane("left", cx),
"j" => self.focus_adjacent_pane("down", cx),
"k" => self.focus_adjacent_pane("up", cx),
"l" => self.focus_adjacent_pane("right", cx),
"h" => self.focus_adjacent_pane("left", window, cx),
"j" => self.focus_adjacent_pane("down", window, cx),
"k" => self.focus_adjacent_pane("up", window, cx),
"l" => self.focus_adjacent_pane("right", window, cx),
"q" => {
if let Some(active_id) = self.active_tab.clone() {
self.close_tab(active_id, cx);
@@ -161,21 +161,19 @@ impl Ashell {
let Some(active_id) = self.active_tab.clone() else {
return;
};
let Some(tab) = self.tabs.iter_mut().find(|t| t.id == active_id) else {
let Some(app_cursor_mode) = self
.tabs
.iter()
.find(|tab| tab.id == active_id)
.map(|tab| tab.app_cursor_mode())
else {
return;
};
if tab.render_snapshot(false).display_offset > 0 {
tab.scroll_to_bottom();
}
tab.clear_selection();
if let Some(bytes) = encode_key(&event.keystroke, tab.app_cursor_mode(), false) {
tab.send_backend(BackendCommand::Input(bytes));
window.prevent_default();
cx.stop_propagation();
cx.notify();
}
let Some(bytes) = encode_key(&event.keystroke, app_cursor_mode, false) else {
return;
};
self.send_terminal_input(bytes, window, cx);
}
pub(crate) fn on_terminal_tab_action(
@@ -200,6 +198,7 @@ impl Ashell {
let Some(active_id) = self.active_tab.clone() else {
return;
};
self.record_ssh_input(&active_id, &bytes);
let Some(tab) = self.tabs.iter_mut().find(|t| t.id == active_id) else {
return;
};
@@ -209,12 +208,33 @@ impl Ashell {
}
tab.clear_selection();
tab.send_backend(BackendCommand::Input(bytes));
let encoded = tab.encode_input(&bytes);
tab.send_backend(BackendCommand::Input(encoded));
window.prevent_default();
cx.stop_propagation();
cx.notify();
}
pub(crate) fn execute_ssh_history_command(
&mut self,
command: String,
window: &mut Window,
cx: &mut Context<Self>,
) {
let is_connected_ssh = self
.active_tab
.as_ref()
.and_then(|active_id| self.tabs.iter().find(|tab| &tab.id == active_id))
.is_some_and(|tab| tab.kind == crate::terminal::TabKind::Ssh && tab.connected);
if !is_connected_ssh {
return;
}
let mut bytes = command.into_bytes();
bytes.push(b'\r');
self.send_terminal_input(bytes, window, cx);
self.close_command_history(cx);
}
pub(crate) fn active_terminal_selection_text(&self) -> Option<String> {
let active_id = self.active_tab.as_ref()?;
self.tabs
@@ -232,6 +252,11 @@ impl Ashell {
let Some(active_id) = self.active_tab.clone() else {
return;
};
let normalized_text = text
.replace('\x1b', "")
.replace("\r\n", "\r")
.replace('\n', "\r");
self.record_ssh_input(&active_id, normalized_text.as_bytes());
let Some(tab) = self.tabs.iter_mut().find(|tab| tab.id == active_id) else {
return;
};
@@ -287,6 +312,8 @@ impl Ashell {
let Some(active_id) = self.active_tab.clone() else {
return;
};
let bytes = text.as_bytes().to_vec();
self.record_ssh_input(&active_id, &bytes);
let Some(tab) = self.tabs.iter_mut().find(|tab| tab.id == active_id) else {
return;
};
@@ -296,11 +323,134 @@ impl Ashell {
}
tab.clear_selection();
self.terminal_marked_text = None;
tab.send_backend(BackendCommand::Input(text.as_bytes().to_vec()));
let encoded = tab.encode_input(&bytes);
tab.send_backend(BackendCommand::Input(encoded));
window.invalidate_character_coordinates();
cx.notify();
}
/// Track the current SSH shell line and persist completed commands.
fn record_ssh_input(&mut self, tab_id: &str, bytes: &[u8]) {
let (session_id, cursor, is_alternate_screen_active) = {
let Some(tab) = self.tabs.iter().find(|tab| {
tab.id == tab_id && tab.kind == crate::terminal::TabKind::Ssh && tab.connected
}) else {
return;
};
let Some(session) = tab.session.as_ref() else {
return;
};
(
session.id.clone(),
tab.cursor_state().map(|cursor| (cursor.row, cursor.col)),
tab.is_alternate_screen_active(),
)
};
if is_alternate_screen_active {
self.ssh_command_buffers.remove(tab_id);
self.ssh_command_starts.remove(tab_id);
return;
}
let submits_command = bytes.iter().any(|byte| matches!(*byte, b'\r' | b'\n'));
let edits_command = bytes
.iter()
.any(|byte| !matches!(*byte, b'\r' | b'\n' | b'\x03'));
if edits_command && !self.ssh_command_starts.contains_key(tab_id) {
if let Some(cursor) = cursor {
self.ssh_command_starts.insert(tab_id.to_string(), cursor);
}
}
let mut rendered_command = if submits_command {
self.ssh_command_starts
.get(tab_id)
.copied()
.and_then(|start| {
self.tabs
.iter()
.find(|tab| tab.id == tab_id)
.map(|tab| tab.render_snapshot(false))
.and_then(|snapshot| terminal_command_text(&snapshot, start))
})
} else {
None
};
let mut completed = Vec::new();
let mut reset_command_start = false;
{
let buffer = self
.ssh_command_buffers
.entry(tab_id.to_string())
.or_default();
let mut in_escape = false;
let mut in_csi = false;
for character in String::from_utf8_lossy(bytes).chars() {
if in_escape {
if character == '[' || character == 'O' {
in_escape = false;
in_csi = true;
} else {
in_escape = false;
}
continue;
}
if in_csi {
if character.is_ascii_alphabetic() || character == '~' {
in_csi = false;
}
continue;
}
match character {
'\x1b' => in_escape = true,
'\r' | '\n' => {
let command =
merge_command_text(rendered_command.take().as_deref(), buffer);
if !command.is_empty() {
completed.push(command);
}
buffer.clear();
reset_command_start = true;
}
'\u{8}' | '\u{7f}' => {
buffer.pop();
}
'\u{15}' => buffer.clear(),
'\u{3}' => {
buffer.clear();
reset_command_start = true;
}
'\u{17}' => {
let trimmed_len = buffer.trim_end().len();
buffer.truncate(trimmed_len);
while let Some((index, character)) = buffer.char_indices().next_back() {
if character.is_whitespace() {
break;
}
buffer.truncate(index);
}
}
character if !character.is_control() => buffer.push(character),
_ => {}
}
}
}
if reset_command_start {
self.ssh_command_starts.remove(tab_id);
}
let mut changed = false;
for command in completed {
changed |= self.config.add_command_history(&session_id, command);
}
if changed {
self.selected_command_history.clear();
self.save_preferences_background();
}
}
pub(crate) fn on_terminal_right_click(
&mut self,
_event: &MouseDownEvent,
@@ -445,15 +595,11 @@ impl Ashell {
if should_scroll {
if row == 0 {
scroll_delta = 2;
} else if row == 1 {
scroll_delta = 1;
} else if row == 2 {
} else if row == 1 || row == 2 {
scroll_delta = 1;
} else if row == max_row {
scroll_delta = -2;
} else if row == max_row.saturating_sub(1) {
scroll_delta = -1;
} else if row == max_row.saturating_sub(2) {
} else if row == max_row.saturating_sub(1) || row == max_row.saturating_sub(2) {
scroll_delta = -1;
}
}
@@ -492,7 +638,7 @@ impl Ashell {
let bounds = self.terminal_bounds.get(active_id)?;
if !bounds.contains(&position) {
// Try other pane bounds
for (_, b) in &self.terminal_bounds {
for b in self.terminal_bounds.values() {
if b.contains(&position) {
// Found a different pane - focus it
// (this path is for click-to-focus; handled via focus_terminal)
@@ -528,7 +674,7 @@ impl Ashell {
// Platform modifier (Cmd on macOS, Ctrl on Windows/Linux) + scroll → zoom terminal font size
if event.modifiers.platform {
let delta = match event.delta {
ScrollDelta::Lines(point) => point.y as f32 * 20.0,
ScrollDelta::Lines(point) => point.y * 20.0,
ScrollDelta::Pixels(point) => point.y.as_f32(),
};
self.terminal_zoom_accumulator += delta;
@@ -629,3 +775,60 @@ impl Ashell {
}
}
}
fn terminal_command_text(
snapshot: &crate::terminal::RenderSnapshot,
start: (usize, usize),
) -> Option<String> {
let logical_lines =
crate::terminal::highlight::build_logical_lines(&snapshot.cells, snapshot.rows);
for line in logical_lines {
if !line.byte_to_cell.iter().any(|(row, _)| *row == start.0) {
continue;
}
let start_byte = line
.byte_to_cell
.iter()
.position(|(row, col)| *row > start.0 || (*row == start.0 && *col >= start.1))?;
let command = line
.text
.get(start_byte..)?
.trim_end_matches(|character: char| character == '\0' || character.is_whitespace())
.replace('\0', "");
if !command.trim().is_empty() {
return Some(command);
}
}
None
}
fn merge_command_text(rendered: Option<&str>, buffered: &str) -> String {
let rendered = rendered.unwrap_or_default().trim();
let buffered = buffered.trim();
if rendered.is_empty() {
return buffered.to_string();
}
if buffered.is_empty() {
return rendered.to_string();
}
if rendered.starts_with(buffered) || rendered.ends_with(buffered) {
return rendered.to_string();
}
if buffered.starts_with(rendered) || buffered.ends_with(rendered) {
return buffered.to_string();
}
let overlap = buffered
.char_indices()
.map(|(index, _)| index)
.chain(std::iter::once(buffered.len()))
.filter(|index| *index > 0 && rendered.ends_with(&buffered[..*index]))
.max()
.unwrap_or(0);
if overlap > 0 {
format!("{rendered}{}", &buffered[overlap..])
} else {
rendered.to_string()
}
}
+388 -28
View File
@@ -3,7 +3,15 @@ pub mod element;
pub mod highlight;
pub mod input;
use std::sync::mpsc::Sender;
use std::{
collections::{HashMap, HashSet},
path::PathBuf,
sync::{
Arc,
atomic::{AtomicU32, Ordering},
mpsc::{SendError, Sender},
},
};
use alacritty_terminal::{
event::{Event, EventListener},
@@ -17,7 +25,8 @@ use gpui::Keystroke;
use crate::session::config::Session;
use crate::sftp::{PreviewData, RemoteEntry};
use crate::system::SystemSnapshot;
use crate::system::{RemotePort, RemoteProcess, SystemSnapshot};
use crate::text_encoding::{StreamingDecoder, TextEncoding};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TabKind {
@@ -31,11 +40,19 @@ pub enum BackendCommand {
Input(Vec<u8>),
Resize { cols: u16, rows: u16 },
SampleMetrics,
SampleProcesses,
SamplePorts,
TerminateProcess { pid: u32 },
Close,
}
#[derive(Debug, Clone)]
pub enum BackendEvent {
Guarded {
current_generation: Arc<AtomicU32>,
generation: u32,
event: Box<BackendEvent>,
},
Output {
tab_id: String,
bytes: Vec<u8>,
@@ -52,6 +69,11 @@ pub enum BackendEvent {
path: String,
entries: Vec<RemoteEntry>,
},
SftpDirectoryFailed {
tab_id: String,
path: String,
reason: String,
},
SftpPreview {
tab_id: String,
preview: PreviewData,
@@ -68,6 +90,31 @@ pub enum BackendEvent {
tab_id: String,
reason: String,
},
RemoteProcesses {
tab_id: String,
processes: Vec<RemoteProcess>,
},
RemoteProcessesUnavailable {
tab_id: String,
reason: String,
},
RemotePorts {
tab_id: String,
ports: Vec<RemotePort>,
},
RemotePortsUnavailable {
tab_id: String,
reason: String,
},
RemoteProcessTerminated {
tab_id: String,
pid: u32,
},
RemoteProcessTerminateFailed {
tab_id: String,
pid: u32,
reason: String,
},
SftpHome {
tab_id: String,
home: String,
@@ -92,14 +139,85 @@ pub enum BackendEvent {
tab_id: String,
title: String,
},
LocalDirectoryChanged {
tab_id: String,
path: std::path::PathBuf,
},
SyncFinished(crate::sync::SyncResult),
}
impl BackendEvent {
pub(crate) fn into_current(self) -> Option<Self> {
match self {
Self::Guarded {
current_generation,
generation,
event,
} if current_generation.load(Ordering::Acquire) == generation => Some(*event),
Self::Guarded { .. } => None,
event => Some(event),
}
}
}
/// Filters events emitted by superseded terminal backends.
///
/// Every backend instance captures the tab generation that was current when it
/// was spawned. Reconnecting advances the shared generation before the old
/// backend is closed, so late events from that backend never reach the UI.
#[derive(Clone)]
pub struct GuardedBackendEventSender {
events: Sender<BackendEvent>,
current_generation: Arc<AtomicU32>,
generation: u32,
}
impl GuardedBackendEventSender {
pub fn new(events: Sender<BackendEvent>) -> Self {
Self {
events,
current_generation: Arc::new(AtomicU32::new(0)),
generation: 0,
}
}
pub fn send(&self, event: BackendEvent) -> Result<(), Box<SendError<BackendEvent>>> {
if self.current_generation.load(Ordering::Acquire) != self.generation {
return Ok(());
}
self.events
.send(BackendEvent::Guarded {
current_generation: self.current_generation.clone(),
generation: self.generation,
event: Box::new(event),
})
.map_err(Box::new)
}
fn next_generation(&self) -> Self {
let generation = self
.current_generation
.fetch_add(1, Ordering::AcqRel)
.wrapping_add(1);
Self {
events: self.events.clone(),
current_generation: self.current_generation.clone(),
generation,
}
}
fn unguarded_sender(&self) -> Sender<BackendEvent> {
self.events.clone()
}
}
#[derive(Clone)]
pub enum BackendTx {
Local(Sender<BackendCommand>),
Ssh(tokio::sync::mpsc::UnboundedSender<BackendCommand>),
Serial(tokio::sync::mpsc::UnboundedSender<BackendCommand>),
/// A restored session that is waiting for the user to confirm reconnecting.
Pending,
}
impl BackendTx {
@@ -114,6 +232,7 @@ impl BackendTx {
Self::Serial(tx) => {
let _ = tx.send(command);
}
Self::Pending => {}
}
}
}
@@ -122,32 +241,32 @@ pub struct TerminalTab {
pub id: String,
pub title: String,
pub dynamic_title: String,
pub terminal_title_received: bool,
pub local_cwd: Option<PathBuf>,
pub kind: TabKind,
pub status: String,
pub connected: bool,
pub disconnected_reason: Option<String>,
/// Incremented each time the tab is reconnected. Used to ignore stale
/// `BackendEvent::Closed` from the previous backend after a retry.
pub backend_generation: u32,
/// Set to `true` when the current backend sends its first `Output` or
/// `Connected` event. Used to skip stale `Closed` events that arrive
/// before the new backend has started producing output.
pub backend_initialized: bool,
pub session: Option<Session>,
text_encoding: TextEncoding,
output_decoder: StreamingDecoder,
processor: Processor,
term: Term<TerminalListener>,
pub cols: u16,
pub rows: u16,
pub backend: std::sync::Arc<std::sync::Mutex<BackendTx>>,
backend_events: GuardedBackendEventSender,
pub scroll_pixel_y: f32,
pub(crate) highlight_cache: std::cell::RefCell<
Option<(
Vec<RenderCell>,
std::collections::HashMap<(i32, i32), gpui::Hsla>,
)>,
>,
pub(crate) highlight_cache: HighlightCache,
}
type HighlightCache = std::cell::RefCell<
Option<(
Vec<RenderCell>,
std::collections::HashMap<(i32, i32), gpui::Hsla>,
)>,
>;
#[derive(Clone, Copy)]
pub struct CursorState {
pub row: usize,
@@ -183,15 +302,218 @@ pub struct ViewportSelection {
pub is_block: bool,
}
#[derive(Clone, Debug)]
pub(crate) struct SftpTreeRow {
pub(crate) path: String,
pub(crate) label: String,
pub(crate) depth: usize,
pub(crate) expanded: bool,
pub(crate) loading: bool,
pub(crate) error: Option<String>,
}
#[derive(Clone, Default)]
pub struct SftpUiState {
pub current_path: String,
pub status: String,
pub entries: Vec<RemoteEntry>,
pub directory_cache: HashMap<String, Vec<RemoteEntry>>,
pub expanded_directories: HashSet<String>,
pub loading_directories: HashSet<String>,
pub directory_errors: HashMap<String, String>,
pub selected_path: Option<String>,
pub preview: Option<PreviewData>,
pub selected_entries: std::collections::HashSet<String>,
pub selected_entries: HashSet<String>,
pub home_dir: String,
pub home_dir_resolved: bool,
}
impl SftpUiState {
pub(crate) fn current_entries(&self) -> &[RemoteEntry] {
self.directory_cache
.get(&self.current_path)
.map(Vec::as_slice)
.unwrap_or(&[])
}
pub(crate) fn begin_directory_load(&mut self, path: &str) {
self.loading_directories.insert(path.to_string());
self.directory_errors.remove(path);
}
pub(crate) fn apply_directory_entries(&mut self, path: String, entries: Vec<RemoteEntry>) {
self.loading_directories.remove(&path);
self.directory_errors.remove(&path);
self.directory_cache.insert(path, entries);
}
pub(crate) fn apply_directory_error(&mut self, path: String, reason: String) {
self.loading_directories.remove(&path);
self.directory_errors.insert(path, reason);
}
pub(crate) fn expand_to(&mut self, path: &str) {
self.expanded_directories
.extend(crate::sftp::remote_path_ancestors(path));
}
pub(crate) fn collapse_all(&mut self) {
self.expanded_directories.clear();
self.expanded_directories.insert("/".to_string());
}
pub(crate) fn tree_rows(&self, show_hidden: bool) -> Vec<SftpTreeRow> {
fn append_rows(
rows: &mut Vec<SftpTreeRow>,
state: &SftpUiState,
path: String,
label: String,
depth: usize,
show_hidden: bool,
) {
let visible_directories = state.directory_cache.get(&path).map(|entries| {
entries
.iter()
.filter(|entry| entry.is_dir && (show_hidden || !entry.name.starts_with('.')))
.cloned()
.collect::<Vec<_>>()
});
let expanded = state.expanded_directories.contains(&path);
rows.push(SftpTreeRow {
path: path.clone(),
label,
depth,
expanded,
loading: state.loading_directories.contains(&path),
error: state.directory_errors.get(&path).cloned(),
});
if !expanded {
return;
}
if let Some(directories) = visible_directories {
for directory in directories {
append_rows(
rows,
state,
directory.full_path,
directory.name,
depth + 1,
show_hidden,
);
}
}
}
let mut rows = Vec::new();
append_rows(
&mut rows,
self,
"/".to_string(),
"/".to_string(),
0,
show_hidden,
);
rows
}
}
#[cfg(test)]
mod sftp_ui_tests {
use super::SftpUiState;
use crate::sftp::RemoteEntry;
fn directory(name: &str, path: &str) -> RemoteEntry {
RemoteEntry {
name: name.to_string(),
full_path: path.to_string(),
is_dir: true,
size: 0,
modified: 0,
}
}
#[test]
fn directory_responses_do_not_replace_the_current_path() {
let mut state = SftpUiState {
current_path: "/home/demo".to_string(),
..SftpUiState::default()
};
state.apply_directory_entries("/var".to_string(), vec![directory("log", "/var/log")]);
assert_eq!(state.current_path, "/home/demo");
assert_eq!(state.directory_cache["/var"][0].full_path, "/var/log");
}
#[test]
fn tree_rows_follow_expansion_and_hidden_directory_preferences() {
let mut state = SftpUiState::default();
state.expanded_directories.insert("/".to_string());
state.expanded_directories.insert("/home".to_string());
state.apply_directory_entries(
"/".to_string(),
vec![
directory(".internal", "/.internal"),
directory("home", "/home"),
],
);
state.apply_directory_entries("/home".to_string(), vec![directory("demo", "/home/demo")]);
let visible_paths = state
.tree_rows(false)
.into_iter()
.map(|row| row.path)
.collect::<Vec<_>>();
assert_eq!(visible_paths, vec!["/", "/home", "/home/demo"]);
let visible_with_hidden = state
.tree_rows(true)
.into_iter()
.map(|row| row.path)
.collect::<Vec<_>>();
assert_eq!(
visible_with_hidden,
vec!["/", "/.internal", "/home", "/home/demo"]
);
}
}
#[cfg(test)]
mod backend_event_tests {
use super::{BackendEvent, GuardedBackendEventSender};
#[test]
fn superseded_backend_events_are_discarded() {
let (events, received) = std::sync::mpsc::channel();
let first = GuardedBackendEventSender::new(events);
first
.send(BackendEvent::Closed {
tab_id: "tab-1".to_string(),
reason: "queued stale close".to_string(),
})
.unwrap();
let second = first.next_generation();
first
.send(BackendEvent::Output {
tab_id: "tab-1".to_string(),
bytes: b"late stale output".to_vec(),
})
.unwrap();
second
.send(BackendEvent::Connected {
tab_id: "tab-1".to_string(),
})
.unwrap();
assert!(received.recv().unwrap().into_current().is_none());
assert!(matches!(
received.recv().unwrap().into_current(),
Some(BackendEvent::Connected { .. })
));
assert!(received.try_recv().is_err());
}
}
impl TerminalTab {
@@ -199,7 +521,7 @@ impl TerminalTab {
id: String,
title: String,
backend: BackendTx,
events: std::sync::mpsc::Sender<BackendEvent>,
backend_events: GuardedBackendEventSender,
) -> Self {
Self::new(
id,
@@ -207,7 +529,7 @@ impl TerminalTab {
TabKind::Local,
"local shell".into(),
backend,
events,
backend_events,
)
}
@@ -215,7 +537,7 @@ impl TerminalTab {
id: String,
session: &Session,
backend: BackendTx,
events: std::sync::mpsc::Sender<BackendEvent>,
backend_events: GuardedBackendEventSender,
) -> Self {
let mut tab = Self::new(
id,
@@ -226,9 +548,10 @@ impl TerminalTab {
session.user, session.host, session.port
),
backend,
events,
backend_events,
);
tab.session = Some(session.clone());
tab.set_text_encoding(session.terminal_encoding);
tab.connected = false;
tab
}
@@ -237,7 +560,7 @@ impl TerminalTab {
id: String,
session: &Session,
backend: BackendTx,
events: std::sync::mpsc::Sender<BackendEvent>,
backend_events: GuardedBackendEventSender,
) -> Self {
let mut tab = Self::new(
id,
@@ -245,7 +568,7 @@ impl TerminalTab {
TabKind::Serial,
format!("connecting serial://{}@{}", session.host, session.baud_rate),
backend,
events,
backend_events,
);
tab.session = Some(session.clone());
tab.connected = false;
@@ -258,32 +581,56 @@ impl TerminalTab {
kind: TabKind,
status: String,
backend: BackendTx,
events: std::sync::mpsc::Sender<BackendEvent>,
backend_events: GuardedBackendEventSender,
) -> Self {
let shared_backend = std::sync::Arc::new(std::sync::Mutex::new(backend));
let events = backend_events.unguarded_sender();
Self {
id: id.clone(),
title: title.clone(),
dynamic_title: title,
terminal_title_received: false,
local_cwd: None,
kind,
status,
connected: matches!(kind, TabKind::Local),
disconnected_reason: None,
backend_generation: 0,
backend_initialized: true,
session: None,
text_encoding: TextEncoding::Utf8,
output_decoder: StreamingDecoder::new(TextEncoding::Utf8),
processor: Processor::new(),
term: new_term(100, 30, shared_backend.clone(), id, events.clone()),
cols: 100,
rows: 30,
backend: shared_backend,
backend_events,
scroll_pixel_y: 0.0,
highlight_cache: std::cell::RefCell::new(None),
}
}
pub fn feed(&mut self, bytes: &[u8]) {
self.processor.advance(&mut self.term, bytes);
let decoded = self.output_decoder.decode(bytes);
self.processor.advance(&mut self.term, &decoded);
}
pub(crate) fn text_encoding(&self) -> TextEncoding {
self.text_encoding
}
pub(crate) fn set_text_encoding(&mut self, encoding: TextEncoding) {
if self.text_encoding == encoding {
return;
}
self.text_encoding = encoding;
self.output_decoder = StreamingDecoder::new(encoding);
if let Some(session) = self.session.as_mut() {
session.terminal_encoding = encoding;
}
}
pub(crate) fn encode_input(&self, bytes: &[u8]) -> Vec<u8> {
self.text_encoding.encode_terminal_input(bytes).into_owned()
}
/// Send a command to the backend. Thread-safe via the shared Arc<Mutex>.
@@ -302,6 +649,15 @@ impl TerminalTab {
}
}
/// Advances this tab to a new backend generation and returns its sender.
/// Call this before closing the old backend so all of its remaining events
/// are discarded immediately.
pub fn advance_backend_events(&mut self) -> GuardedBackendEventSender {
let next = self.backend_events.next_generation();
self.backend_events = next.clone();
next
}
pub fn resize(&mut self, cols: u16, rows: u16) {
let new_cols = cols.max(1);
let new_rows = rows.max(1);
@@ -343,6 +699,10 @@ impl TerminalTab {
self.term.mode().contains(TermMode::APP_CURSOR)
}
pub fn is_alternate_screen_active(&self) -> bool {
self.term.mode().contains(TermMode::ALT_SCREEN)
}
pub fn render_snapshot(&self, keyword_highlight: bool) -> RenderSnapshot {
let rows = self.rows;
let cols = self.cols;
@@ -510,7 +870,7 @@ impl TerminalTab {
if bracketed {
bytes.extend_from_slice(b"\x1b[200~");
}
bytes.extend_from_slice(paste_text.as_bytes());
bytes.extend_from_slice(&self.encode_input(paste_text.as_bytes()));
if bracketed {
bytes.extend_from_slice(b"\x1b[201~");
}
+218
View File
@@ -0,0 +1,218 @@
use std::borrow::Cow;
use encoding_rs::{
BIG5, CoderResult, EUC_KR, Encoding, GB18030, GBK, SHIFT_JIS, UTF_8, UTF_16BE, UTF_16LE,
WINDOWS_1252,
};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum TextEncoding {
#[default]
Utf8,
Gb18030,
Gbk,
Big5,
ShiftJis,
EucKr,
Windows1252,
Utf16Le,
Utf16Be,
}
pub(crate) const TERMINAL_ENCODINGS: &[TextEncoding] = &[
TextEncoding::Utf8,
TextEncoding::Gb18030,
TextEncoding::Gbk,
TextEncoding::Big5,
TextEncoding::ShiftJis,
TextEncoding::EucKr,
TextEncoding::Windows1252,
];
pub(crate) const FILE_ENCODINGS: &[TextEncoding] = &[
TextEncoding::Utf8,
TextEncoding::Utf16Le,
TextEncoding::Utf16Be,
TextEncoding::Gb18030,
TextEncoding::Gbk,
TextEncoding::Big5,
TextEncoding::ShiftJis,
TextEncoding::EucKr,
TextEncoding::Windows1252,
];
impl TextEncoding {
pub(crate) fn label(self) -> &'static str {
match self {
Self::Utf8 => "UTF-8",
Self::Gb18030 => "GB18030",
Self::Gbk => "GBK",
Self::Big5 => "Big5",
Self::ShiftJis => "Shift_JIS",
Self::EucKr => "EUC-KR",
Self::Windows1252 => "Windows-1252",
Self::Utf16Le => "UTF-16 LE",
Self::Utf16Be => "UTF-16 BE",
}
}
pub(crate) fn encoding(self) -> &'static Encoding {
match self {
Self::Utf8 => UTF_8,
Self::Gb18030 => GB18030,
Self::Gbk => GBK,
Self::Big5 => BIG5,
Self::ShiftJis => SHIFT_JIS,
Self::EucKr => EUC_KR,
Self::Windows1252 => WINDOWS_1252,
Self::Utf16Le => UTF_16LE,
Self::Utf16Be => UTF_16BE,
}
}
pub(crate) fn detect_bom(bytes: &[u8]) -> Option<Self> {
if bytes.starts_with(&[0xEF, 0xBB, 0xBF]) {
Some(Self::Utf8)
} else if bytes.starts_with(&[0xFF, 0xFE]) {
Some(Self::Utf16Le)
} else if bytes.starts_with(&[0xFE, 0xFF]) {
Some(Self::Utf16Be)
} else {
None
}
}
pub(crate) fn decode_file(self, bytes: &[u8]) -> (String, bool, bool) {
let bom_len = self.matching_bom_len(bytes);
let (text, had_errors) = self
.encoding()
.decode_without_bom_handling(&bytes[bom_len..]);
(text.into_owned(), had_errors, bom_len > 0)
}
pub(crate) fn encode_file(self, text: &str, with_bom: bool) -> (Vec<u8>, bool) {
if matches!(self, Self::Utf16Le | Self::Utf16Be) {
let mut bytes = Vec::with_capacity(text.len().saturating_mul(2).saturating_add(2));
if with_bom {
bytes.extend_from_slice(if self == Self::Utf16Le {
&[0xFF, 0xFE]
} else {
&[0xFE, 0xFF]
});
}
for code_unit in text.encode_utf16() {
let encoded = if self == Self::Utf16Le {
code_unit.to_le_bytes()
} else {
code_unit.to_be_bytes()
};
bytes.extend_from_slice(&encoded);
}
return (bytes, false);
}
let (encoded, _, had_errors) = self.encoding().encode(text);
let mut bytes = Vec::with_capacity(encoded.len() + 3);
if with_bom {
match self {
Self::Utf8 => bytes.extend_from_slice(&[0xEF, 0xBB, 0xBF]),
Self::Utf16Le => bytes.extend_from_slice(&[0xFF, 0xFE]),
Self::Utf16Be => bytes.extend_from_slice(&[0xFE, 0xFF]),
_ => {}
}
}
bytes.extend_from_slice(&encoded);
(bytes, had_errors)
}
pub(crate) fn default_bom(self) -> bool {
matches!(self, Self::Utf16Le | Self::Utf16Be)
}
pub(crate) fn encode_terminal_input<'a>(self, bytes: &'a [u8]) -> Cow<'a, [u8]> {
if self == Self::Utf8 {
return Cow::Borrowed(bytes);
}
let Ok(text) = std::str::from_utf8(bytes) else {
return Cow::Borrowed(bytes);
};
let (encoded, _, _) = self.encoding().encode(text);
encoded
}
fn matching_bom_len(self, bytes: &[u8]) -> usize {
match self {
Self::Utf8 if bytes.starts_with(&[0xEF, 0xBB, 0xBF]) => 3,
Self::Utf16Le if bytes.starts_with(&[0xFF, 0xFE]) => 2,
Self::Utf16Be if bytes.starts_with(&[0xFE, 0xFF]) => 2,
_ => 0,
}
}
}
pub(crate) struct StreamingDecoder {
decoder: encoding_rs::Decoder,
}
impl StreamingDecoder {
pub(crate) fn new(encoding: TextEncoding) -> Self {
Self {
decoder: encoding.encoding().new_decoder_without_bom_handling(),
}
}
pub(crate) fn decode(&mut self, bytes: &[u8]) -> Vec<u8> {
let initial_capacity = self
.decoder
.max_utf8_buffer_length(bytes.len())
.unwrap_or_else(|| bytes.len().saturating_mul(4).saturating_add(16));
let mut output = String::with_capacity(initial_capacity);
let mut remaining = bytes;
loop {
let (result, read, _) = self.decoder.decode_to_string(remaining, &mut output, false);
remaining = &remaining[read..];
match result {
CoderResult::InputEmpty => break,
CoderResult::OutputFull => {
let additional = self
.decoder
.max_utf8_buffer_length(remaining.len())
.unwrap_or_else(|| remaining.len().saturating_mul(4).saturating_add(16))
.max(16);
output.reserve(additional);
}
}
}
output.into_bytes()
}
}
#[cfg(test)]
mod tests {
use super::{StreamingDecoder, TextEncoding};
#[test]
fn streaming_decoder_preserves_split_multibyte_characters() {
let (encoded, _) = TextEncoding::Gbk.encode_file("中文", false);
let mut decoder = StreamingDecoder::new(TextEncoding::Gbk);
let mut decoded = decoder.decode(&encoded[..1]);
decoded.extend(decoder.decode(&encoded[1..]));
assert_eq!(String::from_utf8(decoded).unwrap(), "中文");
}
#[test]
fn file_encoding_preserves_matching_bom() {
let original = [0xFF, 0xFE, b'A', 0x00];
let (text, had_errors, has_bom) = TextEncoding::Utf16Le.decode_file(&original);
let (encoded, encode_errors) = TextEncoding::Utf16Le.encode_file(&text, has_bom);
assert!(!had_errors);
assert!(!encode_errors);
assert_eq!(encoded, original);
}
}