okooooooo/vps-switch
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: repair the 10 issues found in the full-fleet test

Browse Source 
- exec cancel: send an SSH signal and confirm the process group stopped before reporting cancelled
- terminal: reject pane_id whose host does not match the target host (read/send/key/close/reveal)
- auth: enable login by default on Linux/public bind; keep an explicit user choice
- terminal: build the send baseline from the actual PTY write, not from injection output
- exec: mark stdout/stderr explicitly when output exceeds the 1 MiB cap
- web download: deliver to the browser through the transfer engine and clean up staging
- mcp: unfinished tasks report ok=false; batch inspect fails when any item fails
- local: initialize OS metadata per platform and repair legacy Windows metadata on Linux
- cli config: keep PORT and DATA_DIR in the generated Claude/Codex stdio config
- update check: return 502/ok=false with real stdout/stderr/exitCode on failure
- frontend: add vue-tsc typecheck and fix the standardSQL import
This commit is contained in:
weidu12123 committed 2026-10-16 00:58:31 +08:00
1 parent b836cb0b53
commit 5645b08f22
11 files changed
+870 -283
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend/package.json
+3 -1
View File
@@ -5,6 +5,7 @@
"type": "module",
"scripts": {
"dev": "vite --port 5173",
"typecheck": "vue-tsc --noEmit",
"build": "vite build",
"preview": "vite preview"
},
@@ -31,6 +32,7 @@
"postcss": "^8.4.49",
"tailwindcss": "^3.4.17",
"typescript": "~5.6.2",
"vite": "^6.0.5"
"vite": "^6.0.5",
"vue-tsc": "^3.3.13"
}
}
frontend/pnpm-lock.yaml
Generated
+76
View File
@@ -72,6 +72,9 @@ importers:
vite:
specifier: ^6.0.5
version: 6.4.4(jiti@1.21.7)
vue-tsc:
specifier: ^3.3.13
version: 3.3.13(typescript@5.6.3)
packages:
@@ -495,6 +498,20 @@ packages:
vite: ^5.0.0 || ^6.0.0
vue: ^3.2.25
'@volar/language-core@2.4.28':
resolution: {integrity: sha512-w4qhIJ8ZSitgLAkVay6AbcnC7gP3glYM3fYwKV3srj8m494E3xtrCv6E+bWviiK/8hs6e6t1ij1s2Endql7vzQ==}
'@volar/source-map@2.4.28':
resolution: {integrity: sha512-yX2BDBqJkRXfKw8my8VarTyjv48QwxdJtvRgUpNE5erCsgEUdI2DsLbpa+rOQVAJYshY99szEcRDmyHbF10ggQ==}
'@volar/typescript@2.4.28':
resolution: {integrity: sha512-Ja6yvWrbis2QtN4ClAKreeUZPVYMARDYZl9LMEv1iQ1QdepB6wn0jTRxA9MftYmYa4DQ4k/DaSZpFPUfxl8giw==}
peerDependencies:
typescript: '*'
peerDependenciesMeta:
typescript:
optional: true
'@vue/compiler-core@3.5.43':
resolution: {integrity: sha512-zdiLhnbe1QQqgDT8xZMpNmyqZ3qlI+/Q/FHQco57Kwl/b05HhCzN6eVGN9QU9rbga4CrS0H5SYY8VGHZCt/1Hg==}
@@ -507,6 +524,9 @@ packages:
'@vue/compiler-ssr@3.5.43':
resolution: {integrity: sha512-GF62orf7KiJX9RqrHNGrYBudsQGD0OhJ5nDs90O8UiDuS40+YMYomiXu6w6EuvtXuRDc3MSNis3EaaSKAVSWpg==}
'@vue/language-core@3.3.13':
resolution: {integrity: sha512-axA0Xemn5LuUKuKWob0/kVq7xQ2mJJM6VqA9Exvt35nM9yRC31goWVve/G/Zb2ZCLsXC8FHYjjfXlnshknSCew==}
'@vue/reactivity@3.5.43':
resolution: {integrity: sha512-G/c9GyOZNI2jVaaS6OX1EF1SSFSv7H0ERqNTl4+DTFMlZmB5eVAB53aLQNam/7NL2NPtaDD7RdVrzf8uJzMuOA==}
@@ -540,6 +560,9 @@ packages:
'@xterm/xterm@5.5.0':
resolution: {integrity: sha512-hqJHYaQb5OptNunnyAnkHyM8aCjZ1MEIDTQu1iIbbTD/xops91NB5yq1ZK/dC2JDbVWtF23zUtl9JE2NqwT87A==}
alien-signals@3.2.1:
resolution: {integrity: sha512-I8FjmltrfnDFoZedi5CG8DghVYNhzb/Ijluz7tCSJH0xpd0484Kowhbb1XDYOxfJpU1p5wnM2X54dA+IfGyD1g==}
any-promise@1.3.0:
resolution: {integrity: sha512-7UvmKalWRt1wgjL1RrGxoSJW/0QZFIegpeGvZG9kjp8vrRu55XTHbwnqq2GpXm9uLbcuhxm3IqX9OB4MZR1b2A==}
@@ -718,6 +741,9 @@ packages:
resolution: {integrity: sha512-PXwfBhYu0hBCPw8Dn0E+WDYb7af3dSLVWKi3HGv84IdF4TyFoC0ysxFd0Goxw7nSv4T/PzEJQxsYsEiFCKo2BA==}
engines: {node: '>=8.6'}
muggle-string@0.4.1:
resolution: {integrity: sha512-VNTrAak/KhO2i8dqqnqnAHOa3cYBwXEZe9h+D5h/1ZqFSTEFHdM65lR7RoIqq3tBBYavsOXV84NoHXZ0AkPyqQ==}
mz@2.7.0:
resolution: {integrity: sha512-z81GNO7nnYMEhrGh9LeymoE4+Yr0Wn5McHIZMK5cfQCl+NDX08sCZgUc9/6MHni9IWuFLm1Z3HTCXu2z9fN62Q==}
@@ -742,6 +768,9 @@ packages:
resolution: {integrity: sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==}
engines: {node: '>= 6'}
path-browserify@1.0.1:
resolution: {integrity: sha512-b7uo2UCUOYZcnF/3ID0lulOJi/bafxa1xPe7ZPsammBSpjSWQkjNxlt635YGS2MiR9GjvuXCtz2emr3jbsz98g==}
path-parse@1.0.7:
resolution: {integrity: sha512-LDJzPVEEEPR+y48z93A0Ed0yXb8pAByGWo/k5YYdYgpY2/2EsOsksJrq7lOHxryrVOn1ejG6oAp8ahvOIQD8sw==}
@@ -927,6 +956,15 @@ packages:
yaml:
optional: true
vscode-uri@3.2.0:
resolution: {integrity: sha512-m2gXo3bn0G1kT9InzMf07fTbqMbGtyckj3bH5ktLO+1Ssv+yiATZ4dhwaQv9UZWxJh6E9IFGnQyjgWVDWVBDrg==}
vue-tsc@3.3.13:
resolution: {integrity: sha512-jE4/bXdOMSHVxEGgQCqNllJy+9N3Gx6tfSAsg0yhFK/MAN6eXqISihupnDEKSCt12VVnjN2gI3AqLYcqoh5x/A==}
hasBin: true
peerDependencies:
typescript: '>=5.0.0'
vue@3.5.43:
resolution: {integrity: sha512-o5qZoksdnjIKvW1srZ3ab7pcDNYAerBjRe54D0LBLfRdCYFrSgBHVXokMas35czQc0//lmx4/tuY4ZNQ+Rf2Ng==}
peerDependencies:
@@ -1265,6 +1303,20 @@ snapshots:
vite: 6.4.4(jiti@1.21.7)
vue: 3.5.43(typescript@5.6.3)
'@volar/language-core@2.4.28':
dependencies:
'@volar/source-map': 2.4.28
'@volar/source-map@2.4.28': {}
'@volar/typescript@2.4.28(typescript@5.6.3)':
dependencies:
'@volar/language-core': 2.4.28
path-browserify: 1.0.1
vscode-uri: 3.2.0
optionalDependencies:
typescript: 5.6.3
'@vue/compiler-core@3.5.43':
dependencies:
'@babel/parser': 7.29.9
@@ -1295,6 +1347,16 @@ snapshots:
'@vue/compiler-dom': 3.5.43
'@vue/shared': 3.5.43
'@vue/language-core@3.3.13':
dependencies:
'@volar/language-core': 2.4.28
'@vue/compiler-dom': 3.5.43
'@vue/shared': 3.5.43
alien-signals: 3.2.1
muggle-string: 0.4.1
path-browserify: 1.0.1
picomatch: 4.0.7
'@vue/reactivity@3.5.43':
dependencies:
'@vue/shared': 3.5.43
@@ -1333,6 +1395,8 @@ snapshots:
'@xterm/xterm@5.5.0': {}
alien-signals@3.2.1: {}
any-promise@1.3.0: {}
anymatch@3.1.3:
@@ -1516,6 +1580,8 @@ snapshots:
braces: 3.0.3
picomatch: 2.3.2
muggle-string@0.4.1: {}
mz@2.7.0:
dependencies:
any-promise: 1.3.0
@@ -1532,6 +1598,8 @@ snapshots:
object-hash@3.0.0: {}
path-browserify@1.0.1: {}
path-parse@1.0.7: {}
picocolors@1.1.1: {}
@@ -1717,6 +1785,14 @@ snapshots:
fsevents: 2.3.3
jiti: 1.21.7
vscode-uri@3.2.0: {}
vue-tsc@3.3.13(typescript@5.6.3):
dependencies:
'@volar/typescript': 2.4.28(typescript@5.6.3)
'@vue/language-core': 3.3.13
typescript: 5.6.3
vue@3.5.43(typescript@5.6.3):
dependencies:
'@vue/compiler-dom': 3.5.43
frontend/src/App.vue
+2
View File
@@ -121,6 +121,7 @@ const loadingMcpCalls = ref(false);
const mcpStatus = ref<{
exePath: string;
port: number;
env: Record<string, string>;
claudeCodePath: string;
claudeCodeSynced: boolean;
codexPath: string;
@@ -661,6 +662,7 @@ const displayedMcpConfigJson = computed(() => {
type: 'stdio',
command: mcpStatus.value?.exePath || 'vps-switch.exe',
args: ['--mcp-stdio'],
env: mcpStatus.value?.env,
},
},
},
frontend/src/components/CodeEditor.vue
+2 -2
View File
@@ -14,7 +14,7 @@ import { nginx } from '@codemirror/legacy-modes/mode/nginx';
import { properties } from '@codemirror/legacy-modes/mode/properties';
import { toml } from '@codemirror/legacy-modes/mode/toml';
import { dockerFile } from '@codemirror/legacy-modes/mode/dockerfile';
import { sql } from '@codemirror/legacy-modes/mode/sql';
import { standardSQL } from '@codemirror/legacy-modes/mode/sql';
import { powerShell } from '@codemirror/legacy-modes/mode/powershell';
import { xml } from '@codemirror/legacy-modes/mode/xml';
import { diff } from '@codemirror/legacy-modes/mode/diff';
@@ -155,7 +155,7 @@ function languageFor(filename: string) {
case 'toml':
return StreamLanguage.define(toml);
case 'sql':
return StreamLanguage.define(sql.standardSQL);
return StreamLanguage.define(standardSQL);
case 'xml':
case 'html':
case 'htm':
frontend/src/components/FileBrowserPanel.vue
+14 -2
View File
@@ -124,7 +124,7 @@ let pollTimer: ReturnType<typeof setInterval> | null = null;
// 可点击的面包屑分段：如 /root/123 -> [/] [root] [123]；本机 Windows -> [此电脑] [C:] [Users]
const breadcrumbs = computed<BreadcrumbSegment[]>(() => {
const p = currentPath.value.trim();
const isLocalWin = props.host.kind === 'local';
const isLocalWin = props.host.kind === 'local' && props.host.osInfo?.os === 'windows';
if (isLocalWin) {
const crumbs: BreadcrumbSegment[] = [{ label: '此电脑', path: 'drives' }];
@@ -510,7 +510,19 @@ async function saveFileEditor() {
function triggerDownload(item: FileEntry) {
if (item.isDir) return;
// 改走带断点续传的传输引擎（见 openInlineAction('download')），不再用浏览器直下
if (!['127.0.0.1', 'localhost', '[::1]', '::1'].includes(window.location.hostname)) {
// 公网 Web 的本机是浏览器所在机器，交付给浏览器自身的流式下载链路。
const params = new URLSearchParams({ hostId: props.host.id, path: item.path });
const link = document.createElement('a');
link.href = `/api/files/download?${params}`;
link.download = item.name;
document.body.appendChild(link);
link.click();
link.remove();
ensureTransferPolling();
emit('toast', `已发起下载 ${item.name}，请在浏览器下载列表中查看`);
return;
}
openInlineAction('download', item);
}
frontend/src/env.d.ts
Vendored
+7
View File
@@ -0,0 +1,7 @@
/// <reference types="vite/client" />
declare module '*.vue' {
import type { DefineComponent } from 'vue';
const component: DefineComponent;
export default component;
}
src/api.rs
+166 -158
View File
@@ -1,8 +1,6 @@
use crate::{
db::Database,
files::{
self, join_remote_path, TransferManager,
},
files::{self, join_remote_path, TransferManager},
inspector::InspectorEngine,
mcp::{handle_mcp_request, sync_cli_mcp_configs, ExecTaskManager, JsonRpcRequest},
models::{HostMetaPatch, HostUpsertPayload, ReorderPayload, LOCAL_HOST_ID},
@@ -31,14 +29,13 @@ use futures::{
};
use serde::Deserialize;
use serde_json::json;
#[cfg(windows)]
use std::os::windows::process::CommandExt;
use std::{
collections::HashSet,
sync::{Arc, Mutex},
time::Instant,
};
#[cfg(windows)]
use std::os::windows::process::CommandExt;
use tokio::io::AsyncReadExt;
use tokio_util::io::StreamReader;
use uuid::Uuid;
@@ -60,7 +57,10 @@ pub struct AppState {
pub fn build_api_router(state: Arc<AppState>) -> Router {
let protected_api = Router::new()
.route("/api/system/about", get(api_get_about))
.route("/api/system/autostart", get(api_get_autostart).post(api_set_autostart))
.route(
"/api/system/autostart",
get(api_get_autostart).post(api_set_autostart),
)
.route("/api/system/check-update", post(api_check_update))
.route("/api/mcp/status", get(api_mcp_status).post(api_mcp_sync))
.route("/api/settings", get(api_get_settings))
@@ -383,9 +383,7 @@ async fn list_terminals(
State(state): State<Arc<AppState>>,
Query(query): Query<ListTerminalsQuery>,
) -> impl IntoResponse {
let panes = state
.terminals
.list_panes(query.host_id.as_deref(), false);
let panes = state.terminals.list_panes(query.host_id.as_deref(), false);
(
StatusCode::OK,
Json(json!({
@@ -516,9 +514,9 @@ async fn handle_terminal_ws(socket: WebSocket, pane: Arc<PaneHandle>) {
match incoming {
Some(Ok(Message::Binary(data))) => {
if data.first() == Some(&0x01) && data.len() > 1 {
let _ = pane.cmd_tx.send(PaneCommand::Input(data[1..].to_vec()));
let _ = pane.cmd_tx.send(PaneCommand::Input { bytes: data[1..].to_vec(), sent: None });
} else if !data.is_empty() {
let _ = pane.cmd_tx.send(PaneCommand::Input(data));
let _ = pane.cmd_tx.send(PaneCommand::Input { bytes: data, sent: None });
}
}
Some(Ok(Message::Text(text))) => {
@@ -707,7 +705,14 @@ async fn api_rename_item(
State(state): State<Arc<AppState>>,
Json(p): Json<RenamePayload>,
) -> impl IntoResponse {
match files::rename_item(&state.ssh_pool, &state.db, &p.host_id, &p.old_path, &p.new_path).await
match files::rename_item(
&state.ssh_pool,
&state.db,
&p.host_id,
&p.old_path,
&p.new_path,
)
.await
{
Ok(()) => (StatusCode::OK, Json(json!({ "ok": true }))).into_response(),
Err(e) => err_response(StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
@@ -840,7 +845,15 @@ async fn api_stream_upload(
}
}
/// 浏览器高速流式下载：直接把 64 路并发 SFTP 预读流作为 HTTP 响应输出，同时实时更新传输队列速率
struct TemporaryDownload(std::path::PathBuf);
impl Drop for TemporaryDownload {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
/// 浏览器下载沿用传输引擎，完成校验后通过 HTTP 交付，临时文件随响应关闭而清理。
async fn api_stream_download(
State(state): State<Arc<AppState>>,
Query(q): Query<FileQuery>,
@@ -849,14 +862,6 @@ async fn api_stream_download(
return err_response(StatusCode::BAD_REQUEST, "缺少下载路径").into_response();
};
let host_name = state
.db
.get_record(&q.host_id)
.ok()
.flatten()
.map(|h| h.name)
.unwrap_or_else(|| q.host_id.clone());
let filename = file_path
.rsplit(['/', '\\'])
.next()
@@ -876,82 +881,60 @@ async fn api_stream_download(
return build_download_response(&filename, meta.len(), Body::from_stream(stream));
}
let sftp = match state.ssh_pool.get_or_open_sftp(&state.db, &q.host_id).await {
Ok(s) => s,
let temp_dir =
std::env::temp_dir().join(format!("vps-switch-browser-download-{}", Uuid::new_v4()));
if let Err(e) = tokio::fs::create_dir(&temp_dir).await {
return err_response(StatusCode::INTERNAL_SERVER_ERROR, e).into_response();
}
let cleanup = TemporaryDownload(temp_dir);
let local_path = cleanup.0.join("file");
let task = match files::transfer_local_remote_file(
&state.ssh_pool,
&state.db,
&state.transfers,
&q.host_id,
"download",
&local_path.to_string_lossy(),
&file_path,
)
.await
{
Ok(t) => t,
Err(e) => return err_response(StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
};
let meta = match sftp.metadata(&file_path).await {
Ok(m) => m,
Err(e) => return err_response(StatusCode::NOT_FOUND, format!("{e:#}")).into_response(),
};
if meta.is_dir() {
return err_response(StatusCode::BAD_REQUEST, "不能直接下载文件夹").into_response();
}
let total_size = meta.size.unwrap_or(0);
let remote_file = match sftp.open(&file_path).await {
let file = match tokio::fs::File::open(&local_path).await {
Ok(f) => f,
Err(e) => return err_response(StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
};
let task_id = state.transfers.create_task(
&q.host_id,
&host_name,
"download",
&filename,
None,
&file_path,
total_size,
);
let transfers = state.transfers.clone();
let started = Instant::now();
let stream = futures::stream::unfold(
(remote_file, 0u64, task_id, transfers, false),
move |(mut rfile, mut transferred, tid, tm, done)| async move {
if done {
return None;
}
let mut buf = vec![0u8; 512 * 1024];
match rfile.read(&mut buf).await {
Ok(0) => {
let _ = rfile.close().await;
tm.finish_task(&tid, transferred, started.elapsed().as_secs_f64());
None
}
Ok(n) => {
buf.truncate(n);
transferred += n as u64;
tm.update_progress(
&tid,
transferred,
total_size,
started.elapsed().as_secs_f64(),
);
Some((
Ok::<bytes::Bytes, std::io::Error>(bytes::Bytes::from(buf)),
(rfile, transferred, tid, tm, false),
))
}
Err(err) => {
tm.fail_task(&tid, &err);
Some((Err(err), (rfile, transferred, tid, tm, true)))
}
}
},
);
build_download_response(&filename, total_size, Body::from_stream(stream))
let reader = tokio_util::io::ReaderStream::with_capacity(file, 512 * 1024);
let stream = futures::stream::unfold((reader, cleanup), |(mut reader, cleanup)| async move {
reader.next().await.map(|bytes| (bytes, (reader, cleanup)))
});
build_download_response(&filename, task.transferred_bytes, Body::from_stream(stream))
}
fn build_download_response(filename: &str, size: u64, body: Body) -> Response {
let safe_ascii: String = filename
.chars()
.map(|c| if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { c } else { '_' })
.map(|c| {
if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' {
c
} else {
'_'
}
})
.collect();
let disposition = format!("attachment; filename=\"{safe_ascii}\"");
let encoded: String = filename
.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || b"-._~".contains(&b) {
(b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect();
let disposition = format!("attachment; filename=\"{safe_ascii}\"; filename*=UTF-8''{encoded}");
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/octet-stream")
@@ -1013,8 +996,11 @@ async fn api_create_local_remote_transfer(
Json(p): Json<LocalRemoteTransferPayload>,
) -> impl IntoResponse {
if p.direction != "upload" && p.direction != "download" {
return err_response(StatusCode::BAD_REQUEST, "direction 只能是 upload 或 download")
.into_response();
return err_response(
StatusCode::BAD_REQUEST,
"direction 只能是 upload 或 download",
)
.into_response();
}
if state.db.get_record(&p.host_id).ok().flatten().is_none() {
return err_response(StatusCode::BAD_REQUEST, "主机不存在").into_response();
@@ -1071,7 +1057,10 @@ async fn api_clear_transfers(State(state): State<Arc<AppState>>) -> impl IntoRes
const SESSION_COOKIE_NAME: &str = "vps_switch_session";
fn extract_session_token(headers: &HeaderMap) -> Option<String> {
if let Some(auth) = headers.get(header::AUTHORIZATION).and_then(|v| v.to_str().ok()) {
if let Some(auth) = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
{
if let Some(tok) = auth.strip_prefix("Bearer ") {
if !tok.trim().is_empty() {
return Some(tok.trim().to_string());
@@ -1149,12 +1138,13 @@ async fn api_auth_login(
State(state): State<Arc<AppState>>,
Json(p): Json<LoginPayload>,
) -> Response {
match state
.db
.verify_admin_credentials(&p.username, &p.password)
{
match state.db.verify_admin_credentials(&p.username, &p.password) {
Ok(true) => {
let token = format!("sess_{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
let token = format!(
"sess_{}{}",
Uuid::new_v4().simple(),
Uuid::new_v4().simple()
);
state.sessions.lock().unwrap().insert(token.clone());
let cookie = format!(
"{SESSION_COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age=2592000"
@@ -1178,10 +1168,7 @@ async fn api_auth_login(
}
}
async fn api_auth_logout(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
) -> Response {
async fn api_auth_logout(State(state): State<Arc<AppState>>, headers: HeaderMap) -> Response {
if let Some(tok) = extract_session_token(&headers) {
state.sessions.lock().unwrap().remove(&tok);
}
@@ -1200,7 +1187,12 @@ fn check_autostart_enabled() -> bool {
#[cfg(windows)]
{
std::process::Command::new("reg")
.args(["query", "HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run", "/v", "VPSSwitch"])
.args([
"query",
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run",
"/v",
"VPSSwitch",
])
.creation_flags(CREATE_NO_WINDOW)
.output()
.map(|o| o.status.success())
@@ -1327,62 +1319,72 @@ async fn api_get_about(State(state): State<Arc<AppState>>) -> impl IntoResponse
async fn api_check_update() -> impl IntoResponse {
let current_version = env!("CARGO_PKG_VERSION");
let started = Instant::now();
let mut command = tokio::process::Command::new("gh");
command.args([
"release",
"view",
"--repo",
"weidu12123/vps-switch",
"--json",
"tagName,name,publishedAt,body,url",
]);
#[cfg(windows)]
let gh_res = std::process::Command::new("gh")
.args([
"release",
"view",
"--repo",
"weidu12123/vps-switch",
"--json",
"tagName,name,publishedAt,body,url",
])
.creation_flags(CREATE_NO_WINDOW)
.output();
#[cfg(not(windows))]
let gh_res = std::process::Command::new("gh")
.args([
"release",
"view",
"--repo",
"weidu12123/vps-switch",
"--json",
"tagName,name,publishedAt,body,url",
])
.output();
if let Ok(out) = gh_res {
if out.status.success() {
if let Ok(info) = serde_json::from_slice::<serde_json::Value>(&out.stdout) {
let tag = info["tagName"].as_str().unwrap_or("").trim_start_matches('v');
let has_update = !tag.is_empty() && tag != current_version;
return (
StatusCode::OK,
Json(json!({
"ok": true,
"currentVersion": format!("v{current_version}"),
"latestVersion": info["tagName"].as_str().unwrap_or(&format!("v{current_version}")),
"hasUpdate": has_update,
"publishedAt": info["publishedAt"].as_str(),
"releaseNotes": info["body"].as_str(),
"releaseUrl": info["url"].as_str(),
})),
);
command.creation_flags(CREATE_NO_WINDOW);
let mut stdout = String::new();
let mut stderr = String::new();
let mut exit_code = None;
let error = match command.output().await {
Ok(out) => {
stdout = String::from_utf8_lossy(&out.stdout).into_owned();
stderr = String::from_utf8_lossy(&out.stderr).into_owned();
exit_code = out.status.code();
if out.status.success() {
match serde_json::from_slice::<serde_json::Value>(&out.stdout) {
Ok(info)
if info["tagName"]
.as_str()
.is_some_and(|s| !s.trim().is_empty()) =>
{
let tag = info["tagName"].as_str().unwrap().trim_start_matches('v');
let has_update = !tag.is_empty() && tag != current_version;
return (
StatusCode::OK,
Json(json!({
"ok": true,
"currentVersion": format!("v{current_version}"),
"latestVersion": info["tagName"].as_str().unwrap_or(&format!("v{current_version}")),
"hasUpdate": has_update,
"publishedAt": info["publishedAt"].as_str(),
"releaseNotes": info["body"].as_str(),
"releaseUrl": info["url"].as_str(),
"stdout": stdout,
"stderr": stderr,
"exitCode": exit_code,
"durationMs": started.elapsed().as_millis() as u64,
})),
);
}
Ok(_) => "更新响应缺少版本号".to_string(),
Err(e) => format!("无法解析更新响应: {e}"),
}
} else {
format!("更新查询失败（退出码 {:?}）: {}", exit_code, stderr.trim())
}
}
}
Err(e) => format!("无法运行 gh 查询更新: {e}"),
};
(
StatusCode::OK,
StatusCode::BAD_GATEWAY,
Json(json!({
"ok": true,
"ok": false,
"currentVersion": format!("v{current_version}"),
"latestVersion": format!("v{current_version}"),
"hasUpdate": false,
"message": "当前已是最新版本",
"releaseUrl": "https://github.com/weidu12123/vps-switch",
"error": error,
"stdout": stdout,
"stderr": stderr,
"exitCode": exit_code,
"durationMs": started.elapsed().as_millis() as u64,
})),
)
}
@@ -1432,7 +1434,11 @@ async fn api_update_auth_settings(
{
Ok(settings) => {
// 若刚开启了登录保护，自动为当前管理员签发会话 Cookie，避免改完设置立刻被踢出
let token = format!("sess_{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
let token = format!(
"sess_{}{}",
Uuid::new_v4().simple(),
Uuid::new_v4().simple()
);
state.sessions.lock().unwrap().insert(token.clone());
let cookie = format!(
"{SESSION_COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age=2592000"
@@ -1467,11 +1473,10 @@ async fn api_update_mcp_settings(
State(state): State<Arc<AppState>>,
Json(p): Json<UpdateMcpSettingsPayload>,
) -> impl IntoResponse {
match state.db.update_mcp_settings(
p.mcp_enabled,
p.mcp_require_api_key,
p.regenerate_key,
) {
match state
.db
.update_mcp_settings(p.mcp_enabled, p.mcp_require_api_key, p.regenerate_key)
{
Ok(settings) => (
StatusCode::OK,
Json(json!({
@@ -1519,7 +1524,10 @@ async fn api_update_guard(
}
fn extract_mcp_api_key(headers: &HeaderMap) -> Option<String> {
if let Some(auth) = headers.get(header::AUTHORIZATION).and_then(|v| v.to_str().ok()) {
if let Some(auth) = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
{
if let Some(k) = auth.strip_prefix("Bearer ") {
if !k.trim().is_empty() {
return Some(k.trim().to_string());
@@ -1544,7 +1552,8 @@ async fn api_mcp_rpc(
};
if !settings.mcp_enabled {
return err_response(StatusCode::FORBIDDEN, "MCP Server 当前已在设置中关闭").into_response();
return err_response(StatusCode::FORBIDDEN, "MCP Server 当前已在设置中关闭")
.into_response();
}
// API Key 只在**非回环绑定**（真的对外部署）时才校验。绑在本机回环时，端点只有本机
@@ -1626,4 +1635,3 @@ async fn api_sse_events(State(state): State<Arc<AppState>>) -> impl IntoResponse
});
Sse::new(stream).keep_alive(KeepAlive::new().interval(std::time::Duration::from_secs(15)))
}
src/db.rs
+133 -28
View File
@@ -161,10 +161,10 @@ impl Database {
.or_else(|_| std::env::var("USER"))
.unwrap_or_else(|_| "local".to_string());
let os_info = OsInfo {
os: Some("windows".to_string()),
distro_id: Some("windows".to_string()),
version_id: Some("11".to_string()),
pretty_name: Some("Windows 11 Local".to_string()),
os: Some(std::env::consts::OS.to_string()),
distro_id: Some(std::env::consts::OS.to_string()),
version_id: None,
pretty_name: Some(format!("{} Local", std::env::consts::OS)),
arch: Some(std::env::consts::ARCH.to_string()),
kernel: None,
cpu_model: None,
@@ -191,17 +191,35 @@ impl Database {
now
],
)?;
} else if !cfg!(windows) {
let raw: Option<String> = conn.query_row(
"SELECT os_json FROM hosts WHERE id = ?1",
params![LOCAL_HOST_ID],
|row| row.get(0),
)?;
if let Some(mut os) = raw.and_then(|s| serde_json::from_str::<OsInfo>(&s).ok()) {
if os.os.as_deref() == Some("windows") {
os.os = Some(std::env::consts::OS.to_string());
os.distro_id = Some(std::env::consts::OS.to_string());
os.version_id = None;
os.pretty_name = Some(format!("{} Local", std::env::consts::OS));
os.arch = Some(std::env::consts::ARCH.to_string());
conn.execute(
"UPDATE hosts SET os_json = ?1 WHERE id = ?2",
params![serde_json::to_string(&os)?, LOCAL_HOST_ID],
)?;
}
}
}
Ok(())
}
pub fn count_ssh_hosts(&self) -> Result<usize> {
let conn = self.conn.lock().map_err(|_| anyhow!("数据库锁异常"))?;
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM hosts WHERE kind = 'ssh'",
[],
|row| row.get(0),
)?;
let count: i64 =
conn.query_row("SELECT COUNT(*) FROM hosts WHERE kind = 'ssh'", [], |row| {
row.get(0)
})?;
Ok(count as usize)
}
@@ -212,9 +230,7 @@ impl Database {
let tags: Vec<String> = serde_json::from_str(&tags_raw).unwrap_or_default();
let links: Vec<HostLink> = serde_json::from_str(&links_raw).unwrap_or_default();
let os_info: Option<OsInfo> = os_raw
.as_deref()
.and_then(|s| serde_json::from_str(s).ok());
let os_info: Option<OsInfo> = os_raw.as_deref().and_then(|s| serde_json::from_str(s).ok());
Ok(HostRecord {
id: row.get("id")?,
@@ -284,9 +300,13 @@ impl Database {
pub fn get_record(&self, id: &str) -> Result<Option<HostRecord>> {
let conn = self.conn.lock().map_err(|_| anyhow!("数据库锁异常"))?;
conn.query_row("SELECT * FROM hosts WHERE id = ?1", params![id], Self::row_to_record)
.optional()
.map_err(Into::into)
conn.query_row(
"SELECT * FROM hosts WHERE id = ?1",
params![id],
Self::row_to_record,
)
.optional()
.map_err(Into::into)
}
pub fn get_public_host(&self, id: &str) -> Result<Option<PublicHost>> {
@@ -624,8 +644,9 @@ impl Database {
/// 从本机旧版 1Shell (%APPDATA%\1Shell\data\1shell.db) 导入全部主机与偏好
pub fn import_from_1shell(&self) -> Result<usize> {
let db_path = find_1shell_db_path()
.ok_or_else(|| anyhow!("未找到本机 1Shell 数据库 (%APPDATA%\\1Shell\\data\\1shell.db)"))?;
let db_path = find_1shell_db_path().ok_or_else(|| {
anyhow!("未找到本机 1Shell 数据库 (%APPDATA%\\1Shell\\data\\1shell.db)")
})?;
let src_conn = Connection::open_with_flags(&db_path, OpenFlags::SQLITE_OPEN_READ_ONLY)
.with_context(|| format!("只读打开 1Shell 数据库失败: {}", db_path.display()))?;
@@ -817,8 +838,12 @@ impl Database {
[],
)?;
conn.execute(
"INSERT OR IGNORE INTO settings (key, value) VALUES ('auth_enabled', 'false')",
[],
"INSERT OR IGNORE INTO settings (key, value) VALUES ('auth_enabled', ?1)",
params![if default_auth_enabled() {
"true"
} else {
"false"
}],
)?;
conn.execute(
"INSERT OR IGNORE INTO settings (key, value) VALUES ('admin_username', 'admin')",
@@ -902,17 +927,17 @@ impl Database {
}
pub fn get_app_settings(&self) -> Result<AppSettings> {
let is_vps_mode = !cfg!(windows)
|| std::env::var("HOST")
.map(|h| h == "0.0.0.0")
.unwrap_or(false);
let is_vps_mode = default_auth_enabled();
let auth_enabled = match self.get_setting("auth_enabled")? {
Some(v) => v == "true",
None => {
// Windows 本机默认关闭，VPS/Linux 部署（或绑定 0.0.0.0）默认开启
let default_enabled = is_vps_mode;
let _ = self.set_setting("auth_enabled", if default_enabled { "true" } else { "false" });
self.set_setting(
"auth_enabled",
if default_enabled { "true" } else { "false" },
)?;
default_enabled
}
};
@@ -931,8 +956,8 @@ impl Database {
_ => {
// 若是 VPS 部署且从未设置过密码，自动初始化一个初始密码 "vps-switch"
if is_vps_mode {
if let Ok(Some(enc)) = self.master_key.encrypt("vps-switch") {
let _ = self.set_setting("admin_password_enc", &enc);
if let Some(enc) = self.master_key.encrypt("vps-switch")? {
self.set_setting("admin_password_enc", &enc)?;
true
} else {
false
@@ -1004,7 +1029,10 @@ impl Database {
}
self.set_setting("admin_username", clean_user)?;
self.set_setting("auth_enabled", if final_auth_enabled { "true" } else { "false" })?;
self.set_setting(
"auth_enabled",
if final_auth_enabled { "true" } else { "false" },
)?;
self.get_app_settings()
}
@@ -1052,6 +1080,15 @@ pub struct AppSettings {
pub mcp_api_key: String,
}
fn default_auth_enabled() -> bool {
!cfg!(windows)
|| std::env::var("HOST")
.ok()
.and_then(|host| host.parse::<std::net::IpAddr>().ok())
.map(|ip| !ip.is_loopback())
.unwrap_or(false)
}
fn generate_api_key() -> String {
format!(
"vpsk_{}{}",
@@ -1062,7 +1099,10 @@ fn generate_api_key() -> String {
fn find_1shell_db_path() -> Option<PathBuf> {
if let Ok(appdata) = std::env::var("APPDATA") {
let candidate = PathBuf::from(appdata).join("1Shell").join("data").join("1shell.db");
let candidate = PathBuf::from(appdata)
.join("1Shell")
.join("data")
.join("1shell.db");
if candidate.exists() {
return Some(candidate);
}
@@ -1101,3 +1141,68 @@ fn normalize_tags(tags: Vec<String>) -> Vec<String> {
}
out
}
#[cfg(test)]
mod tests {
use super::*;
fn open_test_database() -> (PathBuf, Database) {
let dir = std::env::temp_dir().join(format!("vps-switch-db-test-{}", Uuid::new_v4()));
let key = MasterKey::load_or_create(&dir).unwrap();
let db = Database::open(&dir, key).unwrap();
(dir, db)
}
#[test]
fn first_start_auth_and_local_metadata_match_platform() {
let (dir, db) = open_test_database();
let settings = db.get_app_settings().unwrap();
if !cfg!(windows) {
assert!(settings.auth_enabled);
assert!(settings.has_admin_password);
assert!(db.verify_admin_credentials("admin", "vps-switch").unwrap());
} else if std::env::var("HOST").is_err() {
assert!(!settings.auth_enabled);
}
let local = db.get_record(LOCAL_HOST_ID).unwrap().unwrap();
assert_eq!(
local.os_info.unwrap().os.as_deref(),
Some(std::env::consts::OS)
);
drop(db);
std::fs::remove_dir_all(dir).unwrap();
}
#[test]
fn existing_auth_choice_survives_restart() {
let (dir, db) = open_test_database();
db.set_setting("auth_enabled", "false").unwrap();
drop(db);
let db = Database::open(&dir, MasterKey::load_or_create(&dir).unwrap()).unwrap();
assert!(!db.get_app_settings().unwrap().auth_enabled);
drop(db);
std::fs::remove_dir_all(dir).unwrap();
}
#[cfg(not(windows))]
#[test]
fn linux_reopen_repairs_legacy_windows_local_metadata() {
let (dir, db) = open_test_database();
db.conn.lock().unwrap().execute(
"UPDATE hosts SET os_json = ?1 WHERE id = ?2",
params![r#"{"os":"windows","distroId":"windows","versionId":"11","prettyName":"Windows 11 Local"}"#, LOCAL_HOST_ID]
).unwrap();
drop(db);
let db = Database::open(&dir, MasterKey::load_or_create(&dir).unwrap()).unwrap();
let os = db
.get_record(LOCAL_HOST_ID)
.unwrap()
.unwrap()
.os_info
.unwrap();
assert_eq!(os.os.as_deref(), Some("linux"));
assert_eq!(os.version_id, None);
drop(db);
std::fs::remove_dir_all(dir).unwrap();
}
}
src/files.rs
+3 -3
View File
@@ -249,7 +249,7 @@ async fn list_local_dir(dir_path: Option<&str>) -> Result<DirListing> {
let raw = dir_path.unwrap_or("").trim();
// Windows 顶层虚拟根目录 "drives"（此电脑）：列出 C:/ D:/ E:/ 等全部可用盘符
if raw.eq_ignore_ascii_case("drives") || raw == "/" || raw == "\\" {
if cfg!(windows) && (raw.eq_ignore_ascii_case("drives") || raw == "/" || raw == "\\") {
let mut drives = Vec::new();
for letter in b'A'..=b'Z' {
let drive_str = format!("{}:/", letter as char);
@@ -276,7 +276,7 @@ async fn list_local_dir(dir_path: Option<&str>) -> Result<DirListing> {
std::env::var("USERPROFILE")
.or_else(|_| std::env::var("HOME"))
.map(PathBuf::from)
.unwrap_or_else(|_| PathBuf::from("C:/"))
.unwrap_or_else(|_| PathBuf::from(if cfg!(windows) { "C:/" } else { "/" }))
} else if raw.len() == 2 && raw.as_bytes()[1] == b':' {
PathBuf::from(format!("{raw}/"))
} else {
@@ -326,7 +326,7 @@ async fn list_local_dir(dir_path: Option<&str>) -> Result<DirListing> {
.parent()
.map(|p| p.to_string_lossy().replace('\\', "/"))
.filter(|p| !p.is_empty() && p != &norm_path)
.or_else(|| Some("drives".to_string()))
.or_else(|| cfg!(windows).then(|| "drives".to_string()))
};
Ok(DirListing {
src/mcp.rs
+373 -55
View File
@@ -84,6 +84,8 @@ pub struct ExecTaskSnapshot {
pub duration_ms: u64,
pub stdout: String,
pub stderr: String,
pub stdout_truncated: bool,
pub stderr_truncated: bool,
pub hint: Option<String>,
pub started_at: String,
}
@@ -120,6 +122,8 @@ impl ExecTaskInner {
duration_ms: dur,
stdout: String::from_utf8_lossy(&self.stdout).to_string(),
stderr: String::from_utf8_lossy(&self.stderr).to_string(),
stdout_truncated: self.stdout.len() > MAX_OUTPUT_BYTES,
stderr_truncated: self.stderr.len() > MAX_OUTPUT_BYTES,
hint: self.hint.clone(),
started_at: self.started_at.clone(),
}
@@ -137,7 +141,7 @@ const MAX_OUTPUT_BYTES: usize = 1024 * 1024;
const MAX_TASKS: usize = 60;
fn append_capped(buf: &mut Vec<u8>, chunk: &[u8]) {
if buf.len() >= MAX_OUTPUT_BYTES {
if buf.len() > MAX_OUTPUT_BYTES || chunk.is_empty() {
return;
}
let room = MAX_OUTPUT_BYTES - buf.len();
@@ -187,17 +191,38 @@ impl ExecTaskManager {
map.values().map(|t| t.lock().unwrap().snapshot()).collect()
}
pub fn cancel_task(&self, task_id: &str) -> Option<ExecTaskSnapshot> {
let arc = self.tasks.lock().unwrap().get(task_id).cloned()?;
let mut g = arc.lock().unwrap();
if let Some(tx) = g.cancel_tx.take() {
let _ = tx.send(());
pub async fn cancel_task(&self, task_id: &str) -> Result<ExecTaskSnapshot> {
let arc = self
.tasks
.lock()
.unwrap()
.get(task_id)
.cloned()
.ok_or_else(|| anyhow!("未找到可取消的命令任务: {task_id}"))?;
{
let mut g = arc.lock().unwrap();
if g.state != "running" {
return Ok(g.snapshot());
}
if let Some(tx) = g.cancel_tx.take() {
let _ = tx.send(());
}
}
if g.state == "running" {
g.state = "cancelled".to_string();
g.ended_ms = Some(g.started_inst.elapsed().as_millis() as u64);
let deadline = Instant::now() + Duration::from_secs(10);
loop {
{
let g = arc.lock().unwrap();
if g.state != "running" {
return Ok(g.snapshot());
}
}
if Instant::now() >= deadline {
return Err(anyhow!(
"取消请求已发送，但尚未确认命令停止（taskId: {task_id}）"
));
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
Some(g.snapshot())
}
pub async fn exec_with_wait(
@@ -369,7 +394,8 @@ async fn run_local_exec_task(
}
tokio::select! {
_ = &mut *cancel_rx => {
let _ = child.kill().await;
let killed = child.kill().await;
finish_exec_cancellation(&inner, killed.err().map(|e| format!("终止本机进程失败: {e}")));
return;
}
res = stdout.read(&mut out_buf), if !out_eof => {
@@ -488,7 +514,14 @@ async fn run_ssh_exec_task(
cancel_rx: &mut oneshot::Receiver<()>,
) {
// 并发执行闸：permit 持有到本任务结束，保证同一主机上的 session channel 不超过 MaxSessions
let _exec_permit = match state.ssh_pool.acquire_exec_slot(&host.id).await {
let slot = tokio::select! {
_ = &mut *cancel_rx => {
finish_exec_cancellation(&inner, None);
return;
}
result = state.ssh_pool.acquire_exec_slot(&host.id) => result,
};
let _exec_permit = match slot {
Ok(p) => p,
Err(e) => {
let mut g = inner.lock().unwrap();
@@ -499,7 +532,14 @@ async fn run_ssh_exec_task(
}
};
let mut channel = match open_host_channel(&state.ssh_pool, &state.db, &host.id).await {
let opened = tokio::select! {
_ = &mut *cancel_rx => {
finish_exec_cancellation(&inner, None);
return;
}
result = open_host_channel(&state.ssh_pool, &state.db, &host.id) => result,
};
let mut channel = match opened {
Ok((_, ch)) => ch,
Err(e) => {
let mut g = inner.lock().unwrap();
@@ -518,12 +558,24 @@ async fn run_ssh_exec_task(
.map(|s| s.eq_ignore_ascii_case("windows"))
.unwrap_or(false);
let pid_prefix = format!("\u{1e}vps-exec:{}:", inner.lock().unwrap().id);
let final_cmd = if is_windows {
wrap_windows_remote_command(cmd)
} else {
cmd.to_string()
// 记录这一条 exec 的进程组，不改用户命令的 shell、输出通道或退出码。
format!(
"printf '\\036vps-exec:{}:%s\\037' \"$$\" >&2; eval '{}'",
inner.lock().unwrap().id,
cmd.replace('\'', "'\\''")
)
};
if cancel_rx.try_recv().is_ok() {
let _ = channel.close().await;
finish_exec_cancellation(&inner, None);
return;
}
if let Err(e) = channel.exec(true, final_cmd.as_str()).await {
let mut g = inner.lock().unwrap();
g.state = "failed".to_string();
@@ -534,11 +586,30 @@ async fn run_ssh_exec_task(
}
let mut exit_code: Option<i32> = None;
let mut exit_signal = None;
let mut cancel_deadline = None;
let mut connection_lost = false;
let mut remote_pid = None;
let mut pid_pending = !is_windows;
let mut pid_buffer = Vec::new();
let mut group_kill_sent = false;
let mut cancel_error = None;
loop {
tokio::select! {
_ = &mut *cancel_rx => {
_ = &mut *cancel_rx, if cancel_deadline.is_none() => {
// CHANNEL_CLOSE 不会终止远端进程。发送 SSH signal 请求，随后等服务端确认退出。
if let Err(e) = channel.signal(russh::Sig::KILL).await {
let _ = channel.close().await;
finish_exec_cancellation(&inner, Some(format!("发送远端终止信号失败，无法确认命令停止: {e}")));
return;
}
cancel_deadline = Some(Instant::now() + Duration::from_secs(5));
}
_ = tokio::time::sleep_until(tokio::time::Instant::from_std(
cancel_deadline.unwrap_or_else(|| Instant::now() + Duration::from_secs(86400))
)), if cancel_deadline.is_some() => {
let _ = channel.close().await;
finish_exec_cancellation(&inner, Some(cancel_error.unwrap_or_else(|| "远端未确认终止信号，无法保证命令已停止".to_string())));
return;
}
msg_opt = channel.wait() => {
@@ -548,18 +619,23 @@ async fn run_ssh_exec_task(
g.last_output_inst = Instant::now();
}
Some(ChannelMsg::ExtendedData { data, .. }) => {
let mut g = inner.lock().unwrap();
let cleaned = if is_windows {
clean_clixml_stderr(&data)
} else if pid_pending {
consume_remote_pid(&mut pid_buffer, &mut pid_pending, &mut remote_pid, &pid_prefix, &data)
} else {
data.to_vec()
};
let mut g = inner.lock().unwrap();
append_capped(&mut g.stderr, &cleaned);
g.last_output_inst = Instant::now();
}
Some(ChannelMsg::ExitStatus { exit_status }) => {
exit_code = Some(exit_status as i32);
}
Some(ChannelMsg::ExitSignal { signal_name, .. }) => {
exit_signal = Some(signal_name);
}
// EOF 只表示"输出结束"：OpenSSH 的 exit-status 常在其之后到达，
// 必须读到 Close 才收尾，否则高频丢退出码（旧构建靠谎报成功掩盖了它）
Some(ChannelMsg::Eof) => {}
@@ -573,6 +649,24 @@ async fn run_ssh_exec_task(
}
}
}
if cancel_deadline.is_some() && !group_kill_sent {
if let Some(pid) = remote_pid {
group_kill_sent = true;
// 某些 OpenSSH 忽略 signal 请求；用同一连接的短控制 channel 终止已确认的进程组。
if let Err(e) = kill_remote_exec_group(state, &host.id, pid).await {
cancel_error = Some(format!("终止远端进程组失败，无法确认命令停止: {e:#}"));
}
}
}
}
if !pid_buffer.is_empty() {
append_capped(&mut inner.lock().unwrap().stderr, &pid_buffer);
}
if cancel_deadline.is_some() && matches!(exit_signal, Some(russh::Sig::KILL)) {
finish_exec_cancellation(&inner, None);
return;
}
let mut g = inner.lock().unwrap();
@@ -613,6 +707,90 @@ async fn run_ssh_exec_task(
}
}
fn consume_remote_pid(
buffer: &mut Vec<u8>,
pending: &mut bool,
pid: &mut Option<u32>,
prefix: &str,
data: &[u8],
) -> Vec<u8> {
buffer.extend_from_slice(data);
if let Some(start) = buffer
.windows(prefix.len())
.position(|w| w == prefix.as_bytes())
{
let value_start = start + prefix.len();
if let Some(end_rel) = buffer[value_start..].iter().position(|b| *b == 0x1f) {
let end = value_start + end_rel;
if let Some(value) = std::str::from_utf8(&buffer[value_start..end])
.ok()
.and_then(|s| s.parse::<u32>().ok())
.filter(|pid| *pid > 1 && *pid <= i32::MAX as u32)
{
*pid = Some(value);
buffer.drain(start..=end);
*pending = false;
return std::mem::take(buffer);
}
}
}
if buffer.len() > 4096 {
*pending = false;
return std::mem::take(buffer);
}
Vec::new()
}
async fn kill_remote_exec_group(state: &Arc<AppState>, host_id: &str, pid: u32) -> Result<()> {
let (_, mut channel) = tokio::time::timeout(
Duration::from_secs(3),
open_host_channel(&state.ssh_pool, &state.db, host_id),
)
.await??;
// /proc 的进程组必须仍是这条 SSH exec 自己的组，不能对共享会话发送组信号。
let command = format!("IFS= read -r s < /proc/{pid}/stat || exit 1; s=${{s##*) }}; set -- $s; [ \"$3\" = \"{pid}\" ] || exit 2; kill -KILL -- -{pid}");
channel.exec(true, command.as_str()).await?;
let mut exit_code = None;
let mut output = Vec::new();
let result = tokio::time::timeout(Duration::from_secs(3), async {
while let Some(msg) = channel.wait().await {
match msg {
ChannelMsg::ExitStatus { exit_status } => exit_code = Some(exit_status),
ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => {
output.extend_from_slice(&data)
}
ChannelMsg::Close => break,
_ => {}
}
}
})
.await;
let _ = channel.close().await;
result?;
if exit_code != Some(0) {
return Err(anyhow!(
"控制命令退出码 {:?}: {}",
exit_code,
String::from_utf8_lossy(&output)
));
}
Ok(())
}
fn finish_exec_cancellation(inner: &Arc<Mutex<ExecTaskInner>>, error: Option<String>) {
let mut g = inner.lock().unwrap();
if g.state == "running" {
g.state = if error.is_some() {
"failed"
} else {
"cancelled"
}
.to_string();
g.hint = error;
g.ended_ms = Some(g.started_inst.elapsed().as_millis() as u64);
}
}
// ==================== MCP JSON-RPC 2.0 协议与 7 大精锐工具定义 ====================
#[derive(Debug, Deserialize)]
@@ -1030,8 +1208,15 @@ async fn dispatch_tool_call(
}
async fn tool_vps_list(state: &Arc<AppState>, args: Value) -> Result<Value> {
let refresh = args.get("refresh").and_then(|v| v.as_bool()).unwrap_or(false);
let role_filter = args.get("role").and_then(|v| v.as_str()).unwrap_or("").trim();
let refresh = args
.get("refresh")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let role_filter = args
.get("role")
.and_then(|v| v.as_str())
.unwrap_or("")
.trim();
let search_q = args
.get("search")
.and_then(|v| v.as_str())
@@ -1167,7 +1352,7 @@ async fn tool_vps_inspect(state: &Arc<AppState>, args: Value) -> Result<Value> {
.await;
Ok(json!({
"ok": true,
"ok": results.iter().all(|r| r["ok"] == true),
"results": results,
}))
}
@@ -1210,7 +1395,9 @@ async fn tool_vps_exec(
}
}
if host_queries.is_empty() {
return Err(anyhow!("请提供 host 或 hosts 参数（直接传主机名，如 'vip-hk'）"));
return Err(anyhow!(
"请提供 host 或 hosts 参数（直接传主机名，如 'vip-hk'）"
));
}
let mut targets = Vec::new();
@@ -1248,6 +1435,8 @@ async fn tool_vps_exec(
duration_ms: 0,
stdout: String::new(),
stderr: format!("{e:#}"),
stdout_truncated: false,
stderr_truncated: false,
hint: None,
started_at: Utc::now().to_rfc3339(),
})
@@ -1288,10 +1477,22 @@ async fn tool_vps_terminal(state: &Arc<AppState>, args: Value) -> Result<Value>
let wait_ms = args.get("wait_ms").and_then(|v| v.as_u64()).unwrap_or(6000);
let pane_id_arg = args.get("pane_id").and_then(|v| v.as_str());
let specified_pane = pane_id_arg
.map(|pid| {
let pane = state
.terminals
.get_pane(pid)
.ok_or_else(|| anyhow!("指定的终端 pane 不存在: {pid}"))?;
if pane.host_id != host.id {
return Err(anyhow!("终端 {pid} 不属于目标主机「{}」", host.name));
}
Ok(pane)
})
.transpose()?;
if action == "close" {
if let Some(pid) = pane_id_arg {
state.terminals.close_pane(&state.ssh_pool, pid).await;
if let Some(pane) = specified_pane {
state.terminals.close_pane(&state.ssh_pool, &pane.id).await;
} else if let Some(pane) = state.terminals.find_active_pane_for_host(&host.id) {
state.terminals.close_pane(&state.ssh_pool, &pane.id).await;
}
@@ -1299,11 +1500,8 @@ async fn tool_vps_terminal(state: &Arc<AppState>, args: Value) -> Result<Value>
}
// 查找或自动创建该主机的终端会话（默认 headless=true 无头运行，除非 reveal_ui=true）
let pane = if let Some(pid) = pane_id_arg {
state
.terminals
.get_pane(pid)
.ok_or_else(|| anyhow!("指定的终端 pane 不存在: {pid}"))?
let pane = if let Some(pane) = specified_pane {
pane
} else if let Some(existing) = state.terminals.find_active_pane_for_host(&host.id) {
existing
} else {
@@ -1323,8 +1521,6 @@ async fn tool_vps_terminal(state: &Arc<AppState>, args: Value) -> Result<Value>
.terminals
.get_pane(&created.id)
.ok_or_else(|| anyhow!("创建终端后获取句柄失败"))?;
// 等待远端初始 shell 提示符就绪
tokio::time::sleep(Duration::from_millis(450)).await;
handle
};
@@ -1346,7 +1542,7 @@ async fn tool_vps_terminal(state: &Arc<AppState>, args: Value) -> Result<Value>
if submit {
bytes.push(b'\r');
}
pane.send_and_wait(bytes, max_lines, wait_ms).await
pane.send_and_wait(bytes, max_lines, wait_ms).await?
}
"key" => {
let raw_input = args
@@ -1375,7 +1571,7 @@ async fn tool_vps_terminal(state: &Arc<AppState>, args: Value) -> Result<Value>
"left" => b"\x1b[D".to_vec(),
other => other.as_bytes().to_vec(),
};
pane.send_and_wait(bytes, max_lines, wait_ms).await
pane.send_and_wait(bytes, max_lines, wait_ms).await?
}
other => return Err(anyhow!("未知的 vps_terminal action: {other}")),
};
@@ -1402,7 +1598,8 @@ async fn tool_vps_fs(state: &Arc<AppState>, args: Value) -> Result<Value> {
match action {
"list" => {
let listing = files::list_dir(&state.ssh_pool, &state.db, &host.id, Some(path), false).await?;
let listing =
files::list_dir(&state.ssh_pool, &state.db, &host.id, Some(path), false).await?;
Ok(json!({
"ok": true,
"host": host.name,
@@ -1448,7 +1645,10 @@ async fn tool_vps_fs(state: &Arc<AppState>, args: Value) -> Result<Value> {
Ok(json!({ "ok": true, "host": host.name, "oldPath": path, "newPath": new_path }))
}
"remove" => {
let is_dir = args.get("is_dir").and_then(|v| v.as_bool()).unwrap_or(false);
let is_dir = args
.get("is_dir")
.and_then(|v| v.as_bool())
.unwrap_or(false);
files::remove_item(&state.ssh_pool, &state.db, &host.id, path, is_dir).await?;
Ok(json!({ "ok": true, "host": host.name, "removed": path }))
}
@@ -1591,7 +1791,10 @@ async fn tool_vps_task(
exec_tasks: &Arc<ExecTaskManager>,
args: Value,
) -> Result<Value> {
let action = args.get("action").and_then(|v| v.as_str()).unwrap_or("list");
let action = args
.get("action")
.and_then(|v| v.as_str())
.unwrap_or("list");
let task_id = args.get("task_id").and_then(|v| v.as_str()).unwrap_or("");
// 阻塞等待秒数：>0 时等到任务结束或超时再返回，**一次调用**即拿到最终结果，
// 取代"查询→running→再查询"的轮询（旧设计里 vps_task 无阻塞模式，是死循环的另一半）。
@@ -1625,7 +1828,7 @@ async fn tool_vps_task(
}
}
}
Ok(json!({ "ok": true, "task": snap }))
Ok(json!({ "ok": snap.state == "completed", "task": snap }))
} else {
let mut t = state
.transfers
@@ -1656,14 +1859,12 @@ async fn tool_vps_task(
}
}
}
Ok(json!({ "ok": true, "transfer": t }))
Ok(json!({ "ok": t.state == "completed", "transfer": t }))
}
}
"cancel" => {
let t = exec_tasks
.cancel_task(task_id)
.ok_or_else(|| anyhow!("未找到可取消的命令任务: {task_id}"))?;
Ok(json!({ "ok": true, "task": t }))
let t = exec_tasks.cancel_task(task_id).await?;
Ok(json!({ "ok": t.state == "cancelled" || t.state == "completed", "task": t }))
}
_ => Ok(json!({
"ok": true,
@@ -1675,7 +1876,11 @@ async fn tool_vps_task(
/// `--mcp-stdio` 本地模式：若常驻主进程正在监听则直接桥接过去（与桌面 UI 共享连接池、探针缓存和终端唤起事件），
/// 若主进程未运行则在当前进程内直接处理。
pub async fn run_stdio_bridge(state: Arc<AppState>, exec_tasks: Arc<ExecTaskManager>, port: u16) -> Result<()> {
pub async fn run_stdio_bridge(
state: Arc<AppState>,
exec_tasks: Arc<ExecTaskManager>,
port: u16,
) -> Result<()> {
let stdin = tokio::io::stdin();
let mut lines = tokio::io::BufReader::new(stdin).lines();
@@ -1766,6 +1971,7 @@ async fn try_forward_to_daemon(port: u16, api_key: &str, raw_json: &str) -> Opti
pub struct CliSyncStatus {
pub exe_path: String,
pub port: u16,
pub env: std::collections::BTreeMap<String, String>,
pub claude_code_path: String,
pub claude_code_synced: bool,
pub codex_path: String,
@@ -1792,12 +1998,22 @@ pub fn sync_cli_mcp_configs(port: u16) -> CliSyncStatus {
let claude_path = home.join(".claude.json");
let codex_path = home.join(".codex").join("config.toml");
let claude_ok = sync_claude_json(&claude_path, &exe_path).unwrap_or(false);
let codex_ok = sync_codex_toml(&codex_path, &exe_path).unwrap_or(false);
let data_dir = std::path::absolute(crate::resolve_data_dir())
.unwrap_or_else(|_| crate::resolve_data_dir());
let env = std::collections::BTreeMap::from([
("PORT".to_string(), port.to_string()),
(
"DATA_DIR".to_string(),
data_dir.to_string_lossy().to_string(),
),
]);
let claude_ok = sync_claude_json(&claude_path, &exe_path, &env).unwrap_or(false);
let codex_ok = sync_codex_toml(&codex_path, &exe_path, &env).unwrap_or(false);
CliSyncStatus {
exe_path,
port,
env,
claude_code_path: claude_path.to_string_lossy().to_string(),
claude_code_synced: claude_ok,
codex_path: codex_path.to_string_lossy().to_string(),
@@ -1805,7 +2021,11 @@ pub fn sync_cli_mcp_configs(port: u16) -> CliSyncStatus {
}
}
fn sync_claude_json(claude_path: &std::path::Path, exe_path: &str) -> Result<bool> {
fn sync_claude_json(
claude_path: &std::path::Path,
exe_path: &str,
env: &std::collections::BTreeMap<String, String>,
) -> Result<bool> {
if !claude_path.exists() {
return Ok(false);
}
@@ -1818,12 +2038,11 @@ fn sync_claude_json(claude_path: &std::path::Path, exe_path: &str) -> Result<boo
let desired = json!({
"type": "stdio",
"command": exe_path,
"args": ["--mcp-stdio"]
"args": ["--mcp-stdio"],
"env": env,
});
let mcp_servers = obj
.entry("mcpServers")
.or_insert_with(|| json!({}));
let mcp_servers = obj.entry("mcpServers").or_insert_with(|| json!({}));
if let Some(servers_obj) = mcp_servers.as_object_mut() {
if servers_obj.get("vps-switch") == Some(&desired) {
return Ok(true);
@@ -1837,21 +2056,23 @@ fn sync_claude_json(claude_path: &std::path::Path, exe_path: &str) -> Result<boo
Ok(true)
}
fn sync_codex_toml(codex_path: &std::path::Path, exe_path: &str) -> Result<bool> {
fn sync_codex_toml(
codex_path: &std::path::Path,
exe_path: &str,
env: &std::collections::BTreeMap<String, String>,
) -> Result<bool> {
if !codex_path.exists() {
return Ok(false);
}
let raw = std::fs::read_to_string(codex_path)?;
let posix_exe = exe_path.replace('\\', "/");
let desired_block = format!(
"[mcp_servers.vps-switch]\ncommand = \"{}\"\nargs = [\"--mcp-stdio\"]",
posix_exe
"[mcp_servers.vps-switch]\ncommand = {}\nargs = [\"--mcp-stdio\"]\nenv = {{ DATA_DIR = {}, PORT = {} }}",
serde_json::to_string(&posix_exe)?,
serde_json::to_string(&env["DATA_DIR"] )?,
serde_json::to_string(&env["PORT"] )?,
);
if raw.contains(&desired_block) {
return Ok(true);
}
let updated = if let Some(start_idx) = raw.find("[mcp_servers.vps-switch]") {
let after_header = start_idx + "[mcp_servers.vps-switch]".len();
let next_section_rel = raw[after_header..].find("\n[");
@@ -1874,7 +2095,104 @@ fn sync_codex_toml(codex_path: &std::path::Path, exe_path: &str) -> Result<bool>
};
let tmp_path = codex_path.with_extension("toml.vps-switch-tmp");
if updated == std::fs::read_to_string(codex_path)? {
return Ok(true);
}
std::fs::write(&tmp_path, updated)?;
std::fs::rename(&tmp_path, codex_path)?;
Ok(true)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn remote_pid_header_handles_split_frames_and_preserves_command_stderr() {
let mut buffer = Vec::new();
let mut pending = true;
let mut pid = None;
let prefix = "\u{1e}vps-exec:test-id:";
assert!(consume_remote_pid(
&mut buffer,
&mut pending,
&mut pid,
prefix,
b"startup warning\n\x1evps-exec:test-id:12"
)
.is_empty());
let output = consume_remote_pid(
&mut buffer,
&mut pending,
&mut pid,
prefix,
b"34\x1fcommand stderr\n",
);
assert_eq!(pid, Some(1234));
assert!(!pending);
assert_eq!(output, b"startup warning\ncommand stderr\n");
}
#[test]
fn invalid_remote_pid_does_not_remove_stderr_or_target_a_process_group() {
let mut buffer = Vec::new();
let mut pending = true;
let mut pid = None;
let prefix = "\u{1e}vps-exec:test-id:";
let mut raw = b"\x1evps-exec:test-id:0\x1f".to_vec();
raw.extend(vec![b'A'; 4096]);
let output = consume_remote_pid(&mut buffer, &mut pending, &mut pid, prefix, &raw);
assert_eq!(output, raw);
assert!(pid.is_none());
assert!(!pending);
}
#[test]
fn output_limit_reports_loss_at_exact_chunk_boundary() {
let mut output = vec![b'A'; MAX_OUTPUT_BYTES];
append_capped(&mut output, &[]);
assert_eq!(output.len(), MAX_OUTPUT_BYTES);
append_capped(&mut output, b"lost bytes");
assert!(String::from_utf8_lossy(&output).ends_with("[输出超过 1 MiB，已截断]"));
let marked_size = output.len();
append_capped(&mut output, b"more lost bytes");
assert_eq!(output.len(), marked_size);
}
#[test]
fn output_limit_marks_crossing_chunk_and_preserves_short_output() {
let mut output = "完整 output".as_bytes().to_vec();
append_capped(&mut output, b"\n");
assert_eq!(String::from_utf8(output).unwrap(), "完整 output\n");
let mut output = vec![b'A'; MAX_OUTPUT_BYTES - 2];
append_capped(&mut output, b"BCDE");
assert_eq!(&output[MAX_OUTPUT_BYTES - 2..MAX_OUTPUT_BYTES], b"BC");
assert!(output.len() > MAX_OUTPUT_BYTES);
}
#[test]
fn cli_configs_keep_daemon_environment_and_neighbor_entries() {
let dir = std::env::temp_dir().join(format!("vps-switch-config-test-{}", Uuid::new_v4()));
std::fs::create_dir_all(&dir).unwrap();
let claude = dir.join("claude.json");
let codex = dir.join("codex.toml");
std::fs::write(&claude, r#"{"mcpServers":{"other":{"command":"keep"}}}"#).unwrap();
std::fs::write(&codex, "model = \"keep\"\n[mcp_servers.vps-switch]\ncommand = \"old\"\nargs = []\n[mcp_servers.other]\ncommand = \"keep\"\n").unwrap();
let env = std::collections::BTreeMap::from([
("PORT".to_string(), "18881".to_string()),
("DATA_DIR".to_string(), "C:\\数据\\quoted\"dir".to_string()),
]);
sync_claude_json(&claude, "C:\\tools\\vps-switch.exe", &env).unwrap();
sync_codex_toml(&codex, "C:\\tools\\vps-switch.exe", &env).unwrap();
let json: Value = serde_json::from_str(&std::fs::read_to_string(&claude).unwrap()).unwrap();
assert_eq!(json["mcpServers"]["other"]["command"], "keep");
assert_eq!(json["mcpServers"]["vps-switch"]["env"], json!(env));
let first = std::fs::read_to_string(&codex).unwrap();
assert!(first.contains("PORT = \"18881\""));
assert!(first.contains(&serde_json::to_string(&env["DATA_DIR"]).unwrap()));
assert!(first.contains("[mcp_servers.other]\ncommand = \"keep\""));
sync_codex_toml(&codex, "C:\\tools\\vps-switch.exe", &env).unwrap();
assert_eq!(std::fs::read_to_string(&codex).unwrap(), first);
std::fs::remove_dir_all(&dir).unwrap();
}
}
src/terminal.rs
+91 -34
View File
@@ -20,7 +20,7 @@ use std::{
sync::{Arc, Mutex},
time::{Duration, Instant},
};
use tokio::sync::{broadcast, mpsc};
use tokio::sync::{broadcast, mpsc, oneshot};
use uuid::Uuid;
const RING_BUFFER_CAPACITY: usize = 2 * 1024 * 1024;
@@ -95,7 +95,10 @@ pub enum PaneBroadcast {
}
pub enum PaneCommand {
Input(Vec<u8>),
Input {
bytes: Vec<u8>,
sent: Option<oneshot::Sender<Result<u64>>>,
},
Paste(String),
Resize(u16, u16),
Close,
@@ -214,8 +217,8 @@ impl PaneInternalState {
(trimmed_screen, cursor_line)
}
fn scan_osc_sequences(&mut self, chunk: &[u8]) -> bool { let mut changed = false;
fn scan_osc_sequences(&mut self, chunk: &[u8]) -> bool {
let mut changed = false;
self.osc_tail.extend_from_slice(chunk);
if self.osc_tail.len() > 4096 {
let keep = 1024;
@@ -500,6 +503,7 @@ pub struct PaneHandle {
pub state: Arc<Mutex<PaneInternalState>>,
pub cmd_tx: mpsc::UnboundedSender<PaneCommand>,
pub broadcast_tx: broadcast::Sender<PaneBroadcast>,
input_lock: tokio::sync::Mutex<()>,
}
impl PaneHandle {
@@ -562,14 +566,22 @@ impl PaneHandle {
bytes: Vec<u8>,
max_lines: usize,
wait_ms: u64,
) -> ScreenSnapshot {
let seq_before = self.state.lock().unwrap().output_seq;
let _ = self.cmd_tx.send(PaneCommand::Input(bytes));
) -> Result<ScreenSnapshot> {
let _input_guard = self.input_lock.lock().await;
let (sent, received) = oneshot::channel();
self.cmd_tx
.send(PaneCommand::Input {
bytes,
sent: Some(sent),
})
.map_err(|_| anyhow::anyhow!("终端输入通道已关闭"))?;
// 基线来自真正写入 PTY 的时刻：初始化/注入输出不能冒充本次输入的响应。
let seq_before = received.await.context("终端在发送输入前已关闭")??;
let deadline = Instant::now() + Duration::from_millis(wait_ms.clamp(200, 60_000));
loop {
tokio::time::sleep(Duration::from_millis(80)).await;
let (exited, seq_now, quiet_ms, running, cursor_line) = {
let (exited, seq_now, quiet_ms, running, integrated, cursor_line) = {
let g = self.state.lock().unwrap();
let (_, cl) = g.read_screen_grid(5);
(
@@ -577,6 +589,7 @@ impl PaneHandle {
g.output_seq,
g.last_output_at.elapsed().as_millis() as u64,
g.running,
g.shell_kind.is_some(),
cl,
)
};
@@ -588,7 +601,9 @@ impl PaneHandle {
if looks_like_interactive_prompt(&cursor_line) && quiet_ms >= 550 {
break;
}
if (!running || looks_like_shell_prompt(&cursor_line)) && quiet_ms >= 350 {
if (!running || (!integrated && looks_like_shell_prompt(&cursor_line)))
&& quiet_ms >= 350
{
break;
}
}
@@ -597,7 +612,7 @@ impl PaneHandle {
}
}
self.read_screen_snapshot(max_lines)
Ok(self.read_screen_snapshot(max_lines))
}
}
@@ -641,7 +656,11 @@ impl TerminalManager {
});
}
pub fn list_panes(&self, host_id_filter: Option<&str>, include_headless: bool) -> Vec<PaneMeta> {
pub fn list_panes(
&self,
host_id_filter: Option<&str>,
include_headless: bool,
) -> Vec<PaneMeta> {
let guard = self.panes.lock().unwrap();
guard
.values()
@@ -700,9 +719,9 @@ impl TerminalManager {
/// 该主机是否还有活着的终端（pane 自己退出也算没了）。
fn has_live_pane(&self, host_id: &str) -> bool {
let guard = self.panes.lock().unwrap();
guard.values().any(|p| {
p.host_id == host_id && !p.state.lock().unwrap().exited
})
guard
.values()
.any(|p| p.host_id == host_id && !p.state.lock().unwrap().exited)
}
pub async fn create_pane(
@@ -809,7 +828,10 @@ fn spawn_local_conpty_pane(
let cmd = {
let mut c = CommandBuilder::new("powershell.exe");
let init_hook = r#"$global:__vps_orig_prompt = $function:prompt; function global:prompt { $ec = if ($?) { 0 } else { if ($LASTEXITCODE) { $LASTEXITCODE } else { 1 } }; $p = ($executionContext.SessionState.Path.CurrentLocation.Path -replace '\\','/'); [Console]::Write("$([char]27)]133;D;$ec$([char]7)$([char]27)]7;file://localhost/$p$([char]7)$([char]27)]133;A$([char]7)"); & $global:__vps_orig_prompt }"#;
c.args(["-NoLogo", "-NoExit", "-Command", init_hook]);
c.args(["-NoLogo", "-NoExit"]);
if headless {
c.args(["-Command", init_hook]);
}
c.env("TERM", "xterm-256color");
c.env("TERM_PROGRAM", "vps-switch");
if let Some(ref cwd) = initial_cwd {
@@ -862,6 +884,7 @@ fn spawn_local_conpty_pane(
state: state.clone(),
cmd_tx,
broadcast_tx: broadcast_tx.clone(),
input_lock: tokio::sync::Mutex::new(()),
});
let state_read = state.clone();
@@ -904,15 +927,32 @@ fn spawn_local_conpty_pane(
let state_cmd = state.clone();
let bcast_cmd = broadcast_tx.clone();
tokio::spawn(async move {
let mut initial_input = true;
while let Some(cmd) = cmd_rx.recv().await {
match cmd {
PaneCommand::Input(bytes) => {
PaneCommand::Input { bytes, sent } => {
if initial_input && sent.is_some() {
if !wait_state(&state_cmd, Duration::from_secs(6), |g| {
prompt_settled(g, 250)
})
.await
{
let _ = sent
.unwrap()
.send(Err(anyhow::anyhow!("终端初始提示符尚未就绪")));
continue;
}
}
initial_input = false;
let seq_before = state_cmd.lock().unwrap().output_seq;
if bytes.contains(&b'\r') || bytes.contains(&b'\n') {
let mut guard = state_cmd.lock().unwrap();
guard.running = true;
}
let _ = writer.write_all(&bytes);
let _ = writer.flush();
let result = writer.write_all(&bytes).and_then(|_| writer.flush());
if let Some(sent) = sent {
let _ = sent.send(result.map(|_| seq_before).map_err(Into::into));
}
}
PaneCommand::Paste(text) => {
let payload = encode_safe_paste(&state_cmd, &text);
@@ -1048,13 +1088,22 @@ async fn inject_shell_integration(
state: &Arc<Mutex<PaneInternalState>>,
) {
/// 还回回显。注入的每一条退出路径都要经过这里。
async fn restore_echo(write: &ChannelWriteHalf<russh::client::Msg>) {
let _ = write.data(&b"stty echo 2>/dev/null\n"[..]).await;
async fn restore_echo(
write: &ChannelWriteHalf<russh::client::Msg>,
state: &Arc<Mutex<PaneInternalState>>,
) {
let seq_before = state.lock().unwrap().output_seq;
if write.data(&b"stty echo 2>/dev/null\n"[..]).await.is_ok() {
let _ = wait_state(state, Duration::from_secs(3), |g| {
g.output_seq > seq_before && prompt_settled(g, 250)
})
.await;
}
}
// P0：等首个提示符。失败（超时/Windows 原生 shell）就把回显还回去、原地退出。
if !wait_state(state, Duration::from_secs(6), |g| prompt_settled(g, 300)).await {
restore_echo(write).await;
restore_echo(write, state).await;
return;
}
let windows_shell = {
@@ -1063,11 +1112,12 @@ async fn inject_shell_integration(
prompt_is_windows_style(&cursor)
};
if windows_shell {
restore_echo(write).await;
restore_echo(write, state).await;
return;
}
// P1：探测 shell 类型（回显已关，这一行在屏幕上不可见）
let seq_before = state.lock().unwrap().output_seq;
if write
.data(format!("{SHELL_PROBE_LINE}\n").as_bytes())
.await
@@ -1075,23 +1125,22 @@ async fn inject_shell_integration(
{
return;
}
let seq_before = state.lock().unwrap().output_seq;
if !wait_state(state, Duration::from_millis(2500), move |g| {
g.output_seq > seq_before && g.shell_kind.is_some()
})
.await
{
restore_echo(write).await;
restore_echo(write, state).await;
return;
}
let Some(shell) = state.lock().unwrap().shell_kind else {
restore_echo(write).await;
restore_echo(write, state).await;
return;
};
// P2：写入钩子。等待条件带 output_seq 护栏——回显已关，上一阶段的提示符仍在
// 光标行上，没有护栏会在注入尚未执行时就误判"已完成"。
let Some(inject_line) = injection_line_for(shell) else {
restore_echo(write).await;
restore_echo(write, state).await;
return;
};
let seq_before_inject = state.lock().unwrap().output_seq;
@@ -1109,7 +1158,7 @@ async fn inject_shell_integration(
// P3：还回回显。注入疑似失败（例如 shell 语法报错）也照样还——报错上下文留在屏幕上
// 由用户自己看，绝不为了"擦干净"去动光标。
restore_echo(write).await;
restore_echo(write, state).await;
}
async fn spawn_ssh_pty_pane(
@@ -1150,6 +1199,7 @@ async fn spawn_ssh_pty_pane(
state: state.clone(),
cmd_tx,
broadcast_tx: broadcast_tx.clone(),
input_lock: tokio::sync::Mutex::new(()),
});
let (mut read_half, write_half) = channel.split();
@@ -1201,7 +1251,11 @@ async fn spawn_ssh_pty_pane(
// 写半边：请求 pty（注入前先把远端回显关掉，注入写进去的每一行都不会留痕）→ 请求 shell
// → 静默注入 shell integration → 进入输入/控制循环。
tokio::spawn(async move {
let modes: &[(Pty, u32)] = if skip_injection { &[] } else { &[(Pty::ECHO, 0)] };
let modes: &[(Pty, u32)] = if skip_injection {
&[]
} else {
&[(Pty::ECHO, 0)]
};
if write_half
.request_pty(
false,
@@ -1226,11 +1280,17 @@ async fn spawn_ssh_pty_pane(
while let Some(cmd) = cmd_rx.recv().await {
match cmd {
PaneCommand::Input(bytes) => {
PaneCommand::Input { bytes, sent } => {
let seq_before = state.lock().unwrap().output_seq;
if bytes.contains(&b'\r') || bytes.contains(&b'\n') {
state.lock().unwrap().running = true;
}
if write_half.data(&bytes[..]).await.is_err() {
let result = write_half.data(&bytes[..]).await;
let failed = result.is_err();
if let Some(sent) = sent {
let _ = sent.send(result.map(|_| seq_before).map_err(Into::into));
}
if failed {
break;
}
}
@@ -1264,10 +1324,7 @@ async fn spawn_ssh_pty_pane(
fn encode_safe_paste(state: &Arc<Mutex<PaneInternalState>>, text: &str) -> Vec<u8> {
let bracketed = state.lock().unwrap().bracketed_paste_enabled();
let sanitized: String = text
.chars()
.filter(|&c| c != '\x1b' && c != '\0')
.collect();
let sanitized: String = text.chars().filter(|&c| c != '\x1b' && c != '\0').collect();
if bracketed {
let mut out = Vec::with_capacity(sanitized.len() + 12);
out.extend_from_slice(b"\x1b[200~");